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 d0d754c635d Acknowledge messages asap in pubsub test (#39183)
d0d754c635d is described below
commit d0d754c635d6ba558a3e001e9070eb9018c1fad1
Author: Shunping Huang <[email protected]>
AuthorDate: Wed Jul 1 10:43:08 2026 -0400
Acknowledge messages asap in pubsub test (#39183)
---
.../apache_beam/io/gcp/pubsub_integration_test.py | 20 +++++++++++---------
1 file changed, 11 insertions(+), 9 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 12b20076457..89fd4461beb 100644
--- a/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py
+++ b/sdks/python/apache_beam/io/gcp/pubsub_integration_test.py
@@ -363,7 +363,6 @@ 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(
@@ -371,8 +370,18 @@ class PubSubIntegrationTest(unittest.TestCase):
'subscription': ordering_sub.name,
'max_messages': 10,
})
+ # Acknowledge received messages immediately. This is critical when
message
+ # ordering is enabled because outstanding (unacknowledged) messages
block
+ # the delivery of subsequent messages with the same ordering key.
+ ack_ids = [msg.ack_id for msg in response.received_messages]
+ if ack_ids:
+ self.sub_client.acknowledge(
+ request={
+ 'subscription': ordering_sub.name,
+ 'ack_ids': ack_ids,
+ })
+
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:
@@ -391,13 +400,6 @@ class PubSubIntegrationTest(unittest.TestCase):
self.assertEqual(received_map[b'order_data001'].ordering_key, 'key1')
self.assertEqual(received_map[b'order_data002'].ordering_key, 'key1')
self.assertEqual(received_map[b'order_data003'].ordering_key, 'key2')
-
- 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})
