[BEAM-8733] Handle the registration request synchronously in the Pyth…

…on SDK harness.

sdks/python/apache_beam/runners/worker/sdk_worker.py

@@ -145,16 +145,12 @@ def _execute(self, task, request):
     self._responses.put(response)
 
   def _request_register(self, request):
-    self._request_execute(request)
+    # registration request is handled synchronously
+    self._execute(
+        lambda: self.create_worker().do_instruction(request), request)
 
   def _request_process_bundle(self, request):
-
-    def task():
-      self._execute(
-          lambda: self.create_worker().do_instruction(request), request)
-    self._worker_thread_pool.submit(task)
-    _LOGGER.debug(
-        "Currently using %s threads." % len(self._worker_thread_pool._workers))
+    self._request_execute(request)
 
   def _request_process_bundle_split(self, request):
     self._request_process_bundle_action(request)
@@ -190,6 +186,8 @@ def task():
           lambda: self.create_worker().do_instruction(request), request)
 
     self._worker_thread_pool.submit(task)
+    _LOGGER.debug(
+        "Currently using %s threads." % len(self._worker_thread_pool._workers))
 
   def create_worker(self):
     return SdkWorker(