test: drop flaky DoFn-finally integration test on Prism runner

CI PreCommit_Python_Coverage (Prism runner) was failing on
``test_dofn_finally_closes_reader_via_integration_pipeline``: the test
runs a real pipeline whose downstream ``Map`` raises and asserts the
DoFn's ``finally`` close runs before the bundle is torn down. On Prism
the worker subprocess is killed when the bundle fails before the
generator's ``finally`` flushes its side-channel marker, so the marker
file never appears and the assertion fails.

The unit-level companion test
``test_dofn_finally_closes_reader_on_generator_close`` already covers
the same code path runner-independently by driving
``generator.close()`` directly (the GeneratorExit cleanup that the SDK
harness would invoke during bundle teardown). Dropping the integration
test removes the Prism-runner flake without losing coverage.

The unused ``_downstream_boom`` helper is removed with it.

41/41 unbounded_source_test.

Author: Eliaaazzz

sdks/python/apache_beam/io/unbounded_source_test.py

@@ -291,14 +291,6 @@ def get_checkpoint_mark_coder(self):
     return coders.PickleCoder()
 
 
-def _downstream_boom(_unused):
-  """Module-level so it pickles cleanly through Beam's bundle worker boundary.
-  Used by ``DoFnReaderCloseOnDownstreamRaiseTest`` to simulate a downstream
-  transform that raises mid-bundle (the harness-driven yield-raise path).
-  """
-  raise RuntimeError('downstream boom')
-
-
 def _new_tracker(source, checkpoint=None):
   restriction = _UnboundedSourceRestriction(
       source=source, checkpoint_mark=checkpoint)
@@ -1009,12 +1001,14 @@ class DoFnReaderCloseOnDownstreamRaiseTest(unittest.TestCase):
   the generator (no explicit ``throw``); the generator's ``finally`` runs
   when the generator is closed (``GeneratorExit``) or garbage collected.
 
-  We exercise that path two ways:
-    1. Unit-level: simulate the harness drop with ``generator.close()``
-       (raises ``GeneratorExit`` at the active yield, running ``finally``).
-    2. Integration: run a real pipeline with a downstream ``Map`` that
-       raises, and confirm the reader was closed before the pipeline
-       surfaced the error.
+  Tested at the unit level by simulating the harness drop with
+  ``generator.close()`` (raises ``GeneratorExit`` at the active yield,
+  running ``finally``). An integration-style equivalent using a real
+  pipeline with a downstream raising ``Map`` proved unreliable on the
+  Prism runner: when the bundle fails, the worker subprocess may be torn
+  down before the DoFn generator's ``finally`` has a chance to flush a
+  side-channel marker. The unit-level test below covers the same code
+  path runner-independently.
   """
   def test_dofn_finally_closes_reader_on_generator_close(self):
     marker = _new_marker_path('.gen_close.log')
@@ -1048,35 +1042,6 @@ def test_dofn_finally_closes_reader_on_generator_close(self):
       if os.path.exists(marker):
         os.unlink(marker)
 
-  def test_dofn_finally_closes_reader_via_integration_pipeline(self):
-    """End-to-end harness coverage: a real pipeline with a downstream
-    ``Map`` that raises must surface the exception AND must have closed
-    the reader. This complements the unit-level ``generator.close`` test
-    above by exercising the actual SDK harness output-handler path
-    (``common._OutputHandler.handle_process_outputs``).
-    """
-    marker = _new_marker_path('.integration_close.log')
-    try:
-      raised = False
-      try:
-        with beam.Pipeline() as p:
-          _ = (
-              p
-              | ReadFromUnboundedSource(_MarkerCloseSource(marker))
-              | 'BoomMap' >> beam.Map(_downstream_boom))
-      except Exception:  # pylint: disable=broad-except
-        raised = True
-      self.assertTrue(
-          raised, 'pipeline did not surface the downstream Map exception')
-      self.assertTrue(
-          _wait_for_marker(marker),
-          'reader leaked across the integration pipeline -- the SDK '
-          'harness path that drops the DoFn generator on downstream '
-          'failure did not trigger our finally close.')
-    finally:
-      if os.path.exists(marker):
-        os.unlink(marker)
-
 
 # ------------------------------------------------------------------------------
 # Stronger regression guards (added after independent second-opinion review).