This is an automated email from the ASF dual-hosted git repository.
shunping 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 f17b370e74a Fix flaky PubSub integration test by deduplicating
messages (#39134)
f17b370e74a is described below
commit f17b370e74a73f5c2edbd6e5a0ed7c12afd7ef98
Author: Shunping Huang <[email protected]>
AuthorDate: Mon Jun 29 10:22:22 2026 -0400
Fix flaky PubSub integration test by deduplicating messages (#39134)
* Deduplicate messages in PubSub ordering key integration test
PubSub guarantees at-least-once delivery, which can result in duplicate
messages being pulled during the test. Deduplicate the received messages
by message ID to prevent assertion failures on message count, while still
acknowledging all received delivery attempts.
* Add comments.
---
.../apache_beam/io/gcp/pubsub_integration_test.py | 22 +++++++++++++++-------
1 file changed, 15 insertions(+), 7 deletions(-)
diff --git a/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py
b/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py
index e67c5f2a370..12b20076457 100644
--- a/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py
+++ b/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py
@@ -362,6 +362,8 @@ class PubSubIntegrationTest(unittest.TestCase):
# Retry pulling to handle PubSub delivery delays
received_messages = []
+ received_message_ids = set()
+ ack_ids = []
deadline = time.time() + 60 # wait up to 60 seconds
while time.time() < deadline:
response = self.sub_client.pull(
@@ -369,7 +371,13 @@ class PubSubIntegrationTest(unittest.TestCase):
'subscription': ordering_sub.name,
'max_messages': 10,
})
- received_messages.extend(response.received_messages)
+ for msg in response.received_messages:
+ ack_ids.append(msg.ack_id)
+ # Pub/Sub guarantees at-least-once delivery, so we must deduplicate
+ # messages by message_id to handle potential duplicate deliveries.
+ if msg.message.message_id not in received_message_ids:
+ received_message_ids.add(msg.message.message_id)
+ received_messages.append(msg)
if len(received_messages) >= len(test_messages):
break
time.sleep(5)
@@ -384,12 +392,12 @@ class PubSubIntegrationTest(unittest.TestCase):
self.assertEqual(received_map[b'order_data002'].ordering_key, 'key1')
self.assertEqual(received_map[b'order_data003'].ordering_key, 'key2')
- ack_ids = [msg.ack_id for msg in received_messages]
- self.sub_client.acknowledge(
- request={
- 'subscription': ordering_sub.name,
- 'ack_ids': ack_ids,
- })
+ if ack_ids:
+ self.sub_client.acknowledge(
+ request={
+ 'subscription': ordering_sub.name,
+ 'ack_ids': ack_ids,
+ })
finally:
self.sub_client.delete_subscription(
request={'subscription': ordering_sub.name})
