From 1ff47bc2f88a8485d9bb2466ba8076ef7bcb6209 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Sat, 27 Jun 2026 04:22:42 -0400 Subject: [PATCH 1/2] 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. --- .../io/gcp/pubsub_integration_test.py | 20 ++++++++++++------- 1 file changed, 13 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 e67c5f2a3708..281536b4e77e 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 @@ def test_batch_write_with_ordering_key(self): # 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,11 @@ def test_batch_write_with_ordering_key(self): '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) + 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 +390,12 @@ def test_batch_write_with_ordering_key(self): 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}) From 1062219c67f080e8729b9cebd531145d838520cb Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Sat, 27 Jun 2026 04:37:41 -0400 Subject: [PATCH 2/2] Add comments. --- sdks/python/apache_beam/io/gcp/pubsub_integration_test.py | 2 ++ 1 file changed, 2 insertions(+) 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 281536b4e77e..12b200764578 100644 --- a/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py +++ b/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py @@ -373,6 +373,8 @@ def test_batch_write_with_ordering_key(self): }) 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)