Revert "Enable prism on faild pipelines and rebench."

This reverts commit bc648d5.

Author: shunping

sdks/python/apache_beam/examples/complete/game/leader_board_test.py

@@ -65,10 +65,7 @@ def test_leader_board_users(self):
       result = (
           self.create_data(p)
           | leader_board.CalculateUserScores(allowed_lateness=120))
-      assert_that(
-          result,
-          equal_to([('user2_team2', 2), ('user3_team3', 8), ('user4_team3', 5),
-                    ('user1_team1', 50)]))
+      assert_that(result, equal_to([]))
 
 
 if __name__ == '__main__':

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

@@ -124,11 +124,11 @@ def accept(self, pipeline, is_interactive):
         # double fires on AfterCount trigger, once appropriately, and once
         # incorrectly at the end of the window. This if condition could be
         # more targeted, but for now we'll just ignore all unsafe triggers.
-        #if pipeline.allow_unsafe_triggers:
-        #  self.supported_by_prism_runner = False
+        if pipeline.allow_unsafe_triggers:
+          self.supported_by_prism_runner = False
         # TODO(https://github.com/apache/beam/issues/33623): Prism currently
         # does not support interactive mode
-        if is_in_ipython() or is_interactive:
+        elif is_in_ipython() or is_interactive:
           self.supported_by_prism_runner = False
         # TODO(https://github.com/apache/beam/issues/33623): Prism currently
         # does not support the update compat flag
@@ -144,10 +144,6 @@ def accept(self, pipeline, is_interactive):
           # It does sometimes, but at volume suites start to fail. We will try
           # to enable this in a future release.
           self.supported_by_prism_runner = False
-
-        _LOGGER.error(
-            'PrismRunnerSupportVisotr returns' +
-            str(self.supported_by_prism_runner))
         return self.supported_by_prism_runner
 
       def visit_transform(self, applied_ptransform):
@@ -204,18 +200,13 @@ def visit_transform(self, applied_ptransform):
             transform.get_windowing('').windowfn, beam.window.Sessions):
           self.supported_by_prism_runner = False
 
-        if not self.supported_by_prism_runner:
-          _LOGGER.error(
-              'Visit ' + str(applied_ptransform) + ' returns ' +
-              str(self.supported_by_prism_runner))
-
     # Use BundleBasedDirectRunner if other runners are missing needed features.
     runner = BundleBasedDirectRunner()
 
     # Check whether all transforms used in the pipeline are supported by the
     # PrismRunner
     if _PrismRunnerSupportVisitor().accept(pipeline, self._is_interactive):
-      _LOGGER.error('Running pipeline with PrismRunner.')
+      _LOGGER.info('Running pipeline with PrismRunner.')
       from apache_beam.runners.portability import prism_runner
       runner = prism_runner.PrismRunner()
 
@@ -225,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.error(
+          _LOGGER.info(
               'Pipeline failed on PrismRunner, falling back to DirectRunner.')
           runner = BundleBasedDirectRunner()
         else:
@@ -234,14 +225,13 @@ 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.error('Exception with PrismRunner:\n %s\n' % (e))
-        _LOGGER.error('Falling back to DirectRunner')
+        _LOGGER.info('Exception with PrismRunner:\n %s\n' % (e))
+        _LOGGER.info('Falling back to DirectRunner')
         runner = BundleBasedDirectRunner()
 
     # Check whether all transforms used in the pipeline are supported by the
     # FnApiRunner, and the pipeline was not meant to be run as streaming.
     if _FnApiRunnerSupportVisitor().accept(pipeline):
-      _LOGGER.error('Running pipeline with FnApiRunner.')
       from apache_beam.portability.api import beam_provision_api_pb2
       from apache_beam.runners.portability.fn_api_runner import fn_runner
       from apache_beam.runners.portability.portable_runner import JobServiceHandle

sdks/python/apache_beam/transforms/trigger_test.py

@@ -663,8 +663,7 @@ def format_result(k, vs):
           equal_to(
               list({
                   'A-5': {1, 2, 3, 4, 5},
-                  # A-10, A-11 emit on the closing of window [10, 19]
-                  'A-2': {10, 11},
+                  # A-10, A-11 never emitted due to AfterCount(3) never firing.
                   'B-4': {6, 7, 8, 9},
                   'B-3': {10, 15, 16},
               }.items())))