This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 857dfc9a875 [mqtt] Fix streaming xlang IT failing the Messaging
PostCommit (#39088)
857dfc9a875 is described below
commit 857dfc9a87522f435949f118fff5271b3838befa
Author: Tobias Kaymak <[email protected]>
AuthorDate: Wed Jun 24 15:44:06 2026 +0200
[mqtt] Fix streaming xlang IT failing the Messaging PostCommit (#39088)
* [mqtt] Fix streaming xlang IT: amend test pipeline options, run
non-blocking
---
...m_PostCommit_Python_Xlang_Messaging_Direct.json | 2 +-
.../io/external/xlang_mqttio_it_test.py | 23 +++++++++++++++-------
2 files changed, 17 insertions(+), 8 deletions(-)
diff --git
a/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
b/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
index e3d6056a5de..c537844dc84 100644
--- a/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
+++ b/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run",
- "modification": 1
+ "modification": 3
}
diff --git a/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
b/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
index fa6ebed06ef..889096d1f22 100644
--- a/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
+++ b/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
@@ -33,7 +33,7 @@ import unittest
import pytest
import apache_beam as beam
-from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.options.pipeline_options import PortableOptions
from apache_beam.options.pipeline_options import StandardOptions
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.typehints.row_type import RowTypeConstraint
@@ -214,12 +214,21 @@ class CrossLanguageMqttIOTest(unittest.TestCase):
publisher.start()
subscriber.start()
- options = PipelineOptions([
- '--runner=PrismRunner',
- '--environment_type=LOOPBACK',
- '--streaming',
- ])
- p = TestPipeline(options=options)
+ # MqttIO read is unbounded, so this pipeline runs in streaming mode and
+ # never terminates on its own. Amend the harness-provided pipeline options
+ # rather than discarding them: enable streaming, run non-blocking so the
+ # observe-then-cancel logic below can execute, and target the Prism
portable
+ # runner. The latter is required because SwitchingDirectRunner disables its
+ # Prism delegation for pipelines containing external (cross-language)
+ # transforms (see runners/direct/direct_runner.py) and falls back to the
+ # BundleBasedDirectRunner, which cannot execute an unbounded read.
+ # The runner is instantiated during TestPipeline construction, so it must
be
+ # passed to the constructor; the remaining harness-provided options are
+ # preserved and only amended (streaming + LOOPBACK environment) afterwards.
+ p = TestPipeline(runner='PrismRunner', blocking=False)
+ p.get_pipeline_options().view_as(StandardOptions).streaming = True
+ p.get_pipeline_options().view_as(
+ PortableOptions).environment_type = 'LOOPBACK'
p.not_use_test_runner_api = True
_ = (
p
