Use Java expansion services as workers by default in Python.

Due to the AnyOf environment, remote runners can choose more
expensive but remote-friendly options such as docker.

CHANGES.md

@@ -71,6 +71,9 @@
 
 * X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * The Python SDK now type checks `collections.abc.Collections` types properly. Some type hints that were erroneously allowed by the SDK may now fail. ([#29272](https://github.com/apache/beam/pull/29272))
+* Running multi-language pipelines locally no longer requires Docker.
+  Instead, the same (generally auto-started) subprocess used to perform the
+  expansion can also be used as the cross-language worker.
 
 ## Breaking Changes
 

sdks/python/apache_beam/transforms/external.py

@@ -980,7 +980,12 @@ def _default_args(self):
     to_stage = ','.join([self._path_to_jar] + sum((
         JavaJarExpansionService._expand_jars(jar)
         for jar in self._classpath or []), []))
-    return ['{{PORT}}', f'--filesToStage={to_stage}']
+    args = ['{{PORT}}', f'--filesToStage={to_stage}']
+    # TODO(robertwb): See if it's possible to scope this per pipeline.
+    # Checks to see if the cache is being used for this server.
+    if subprocess_server.SubprocessServer._cache._live_owners:
+      args.append('--alsoStartLoopbackWorker')
+    return args
 
   def __enter__(self):
     if self._service_count == 0: