(beam) branch master updated: Acknowledge messages asap in pubsub test (#39183)

[shunping]

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 <shunp...@google.com>
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})