This is an automated email from the ASF dual-hosted git repository. Abacn pushed a commit to branch pr-39572 in repository https://gitbox.apache.org/repos/asf/beam.git
commit f274b1b35a14123afb17547cae9e2e5683864ab2 Author: Yi Hu <[email protected]> AuthorDate: Wed Aug 19 11:56:55 2026 -0400 [debug only] check if test fixed --- .../apache_beam/io/external/xlang_jmsio_it_test.py | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py b/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py index c1922eb26ba..ff231b7d566 100644 --- a/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py +++ b/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py @@ -92,7 +92,8 @@ class _BaseJmsIOTest(unittest.TestCase): subscriber_result = {} def publish(): - self.produce(source_queue, remaining_records) + time.sleep(5) # debug only: delibrately delay element written + self.produce(source_queue, NUM_RECORDS) stop_event = threading.Event() @@ -110,12 +111,6 @@ class _BaseJmsIOTest(unittest.TestCase): break _LOGGER.info('received %s messages', len(received_messages)) - # TODO(https://github.com/apache/beam/issues/39446): Clean up - # pre-publishing Prism runner issue resolved - initial_records = 10 - remaining_records = NUM_RECORDS - initial_records - self.produce(source_queue, initial_records) - publisher = threading.Thread(target=publish, daemon=True) subscriber = threading.Thread(target=subscribe, daemon=True) @@ -143,11 +138,11 @@ class _BaseJmsIOTest(unittest.TestCase): connection_param), queue=sink_queue, expansion_service=self.expansion_service)) - publisher.start() result = p.run() + publisher.start() subscriber.start() try: - subscriber.join(timeout=20) # 1.5 min + subscriber.join(timeout=20) finally: stop_event.set() publisher.join() @@ -160,8 +155,7 @@ class _BaseJmsIOTest(unittest.TestCase): received = subscriber_result.get('received', []) self.assertEqual(len(received), NUM_RECORDS) - # there are identical records - self.assertEqual(len(set(received)), NUM_RECORDS - initial_records) + self.assertEqual(len(set(received)), NUM_RECORDS) class ActiveMQJmsIOTest(_BaseJmsIOTest):
