Add logging msg in SwitchingDirectRunner

Author: shunping

sdks/python/apache_beam/runners/direct/direct_runner.py

@@ -206,7 +206,7 @@ def visit_transform(self, applied_ptransform):
     # Check whether all transforms used in the pipeline are supported by the
     # PrismRunner
     if _PrismRunnerSupportVisitor().accept(pipeline, self._is_interactive):
-      _LOGGER.info('Running pipeline with PrismRunner.')
+      _LOGGER.exception('Running pipeline with PrismRunner.')
       from apache_beam.runners.portability import prism_runner
       runner = prism_runner.PrismRunner()
 
@@ -216,7 +216,7 @@ def visit_transform(self, applied_ptransform):
         # probably failed on job submission.
         if (PipelineState.is_terminal(pr.state) and
             pr.state != PipelineState.DONE):
-          _LOGGER.info(
+          _LOGGER.exception(
               'Pipeline failed on PrismRunner, falling back to DirectRunner.')
           runner = BundleBasedDirectRunner()
         else:
@@ -225,8 +225,8 @@ def visit_transform(self, applied_ptransform):
         # If prism fails in Preparing the portable job, then the PortableRunner
         # code raises an exception. Catch it, log it, and use the Direct runner
         # instead.
-        _LOGGER.info('Exception with PrismRunner:\n %s\n' % (e))
-        _LOGGER.info('Falling back to DirectRunner')
+        _LOGGER.exception('Exception with PrismRunner:\n %s\n' % (e))
+        _LOGGER.exception('Falling back to DirectRunner')
         runner = BundleBasedDirectRunner()
 
     # Check whether all transforms used in the pipeline are supported by the
@@ -240,6 +240,7 @@ def visit_transform(self, applied_ptransform):
       provision_info = fn_runner.ExtendedProvisionInfo(
           beam_provision_api_pb2.ProvisionInfo(
               pipeline_options=encoded_options))
+      _LOGGER.exception("Use FnApiRunner")
       runner = fn_runner.FnApiRunner(provision_info=provision_info)
 
     return runner.run_pipeline(pipeline, options)

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

@@ -472,6 +472,8 @@ def __init__(self, data_buffer_time_limit_ms=0):
     self._closed = False
     self._exception = None  # type: Optional[Exception]
 
+    _LOGGER.exception("start grpc data channel")
+
   def close(self):
     # type: () -> None
     self._to_send.put(self._WRITES_FINISHED)