pubsub: acquire FlowController before releasing by pongad · Pull Request #1831 · googleapis/google-cloud-java

pongad · 2017-03-30T07:27:25Z

No description provided.

coveralls · 2017-03-30T07:35:57Z

Changes Unknown when pulling 02c617e on pongad:sub-flake into ** on GoogleCloudPlatform:master**.

Sign in to view

  public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> responseMessages) {
-    int receivedMessagesCount = responseMessages.size();
-    if (receivedMessagesCount == 0) {
+    if (responseMessages.size() == 0) {


Sign in to view

+    try {
+      flowController.reserve(responseMessages.size(), totalByteCount);
+    } catch (FlowController.FlowControlException unexpectedException) {
+      throw new IllegalStateException("Flow control unexpected exception", unexpectedException);


Sign in to view

+    }
    messagesWaiter.incrementPendingMessages(responseMessages.size());
+
    Iterator<AckHandler> acksIterator = ackHandlers.iterator();


pongad · 2017-04-04T08:05:59Z

Going through this the second revealed more bugs. I shuffled around a couple more things and documented some happen-before edges that we need to observe.

@garrettjonesgoogle PTAL
@davidtorres could you please take a look?

coveralls · 2017-04-04T08:20:56Z

Coverage decreased (-0.004%) to 80.957% when pulling 28d348d on pongad:sub-flake into c42d5d3 on GoogleCloudPlatform:master.

Sign in to view

-    synchronized (outstandingAckHandlers) {
-      outstandingAckHandlers.add(
-          new ExtensionJob(expiration, INITIAL_ACK_DEADLINE_EXTENSION_SECONDS, ackHandlers));
+  private static int totalMessageSize(Collection<ReceivedMessage> messages)  {


Sign in to view

-      int messageSize = pubsubMessage.getMessage().getSerializedSize();
-      totalByteCount += messageSize;
-      ackHandlers.add(new AckHandler(pubsubMessage.getAckId(), messageSize));
+      ackHandlers.add(new AckHandler(pubsubMessage.getAckId(), pubsubMessage.getMessage().getSerializedSize()));


Sign in to view

+      // AckDeadlineAlarm modifies lists in outstandingAckHandlers in-place and might run at any time.
+      // We will also later iterate over ackHandlers when we give messages to user code.
+      // We must create a new list to pass to outstandingAckHandlers,
+      // so that we can't iterate and modify the list concurrently.


Sign in to view

    messagesWaiter.incrementPendingMessages(responseMessages.size());

+    // Reserving flow control must happen before we give the messages to the user,
+    // otherwise the user code might be given too many messages to process at once.


Sign in to view

+    setupNextAckDeadlineExtensionAlarm(expiration);
+
+    // Deadline extension must be set up before we reserve flow control.
+    // Flow control might block for a while, and extension will keep messages from expiring.


coveralls · 2017-04-05T03:11:16Z

Coverage decreased (-0.2%) to 80.752% when pulling d08b5f6 on pongad:sub-flake into c42d5d3 on GoogleCloudPlatform:master.

pongad · 2017-04-05T03:13:47Z

@garrettjonesgoogle Thank you for the rigorous review. PTAL.

garrettjonesgoogle

@pongad thank you for your thoroughness in bug fixing! LGTM.

This reverts commit 3717ac6.

This reverts commit 3717ac6. This change brings up another serious bug. If the number of messages we pull in one RPC is greater than the number size of the semaphore, we deadlock forever. Will redo this later.

@test

* updating 'PubSubExample' to latest api (#1808) * updating pubsub sample to latest api * Pubsub update (#1818) * Update GAPIC layer * Manual updates to support pubsub changes * Update spi classes (#1817) * Update README.md (#1825) [ci skip] * Add DURABLE_REDUCED_AVAILABILITY storage class (#1834) Otherwise we crash when servers return this value. * logging: make flush wait for writes (#1815) This PR still isn't completely correct, since it does not force any RPC to immediately be issued. However, flush should now correctly wait for RPCs representing prior calls to publish to complete and any failures to be reported to ErrorManager before returning. * use new PartitionKey implementation (#1841) bumping gax version to 0.8.0 for the new implementation. reflects googleapis/gapic-generator#1153 . * Release 0.11.0 * Updating version in README files. [ci skip] * Update version to 0.11.1-SNAPSHOT (#1843) * Added a missing @test annotation (#1842) * pubsub: make Subscriber use ApiService (#1824) Fixes #1761. * SPI: Adding @experimentalapi back to logging client classes (#1844) * Bumping NL, Translate to beta (#1848) * Release 0.11.1 * Updating version in README files. [ci skip] * Update version to 0.11.2-SNAPSHOT (#1852) * pubsub: remove obsolete doc references (#1823) This PR removes references to the deprecated code of the docs. The emulator section is rewritten. Fixes #1789. * adding functions to manage life cycle of resources in ITComputeTest (#1768) *adding functions to ITComputeTest, in order to make sure resources created during tests can be properly deleted even if tests fail or become timed out. * refactor and incorporate feedbacks * implement `add` function with compile-time type checking * add `remove` method to remove a resource from managed resources * use Id's as handles to resources * fix copyright header * rename class name * modify remove function, pass delete function to each add method * address comments * Add Speech v1. (#1858) * Make logging overrides the default channel provider (#1820) * Added more unit tests for SessionPool (#1862) * Rename Translate title to Translation [ci skip] (#1867) * Release 0.11.2 * Updating version in README files. [ci skip] * Update pom.xml version to 0.11.3-SNAPSHOT (#1870) * pubsub: acquire FlowController before releasing (#1831) * Revert "pubsub: acquire FlowController before releasing (#1831)" (#1872) This reverts commit 3717ac6. This change brings up another serious bug. If the number of messages we pull in one RPC is greater than the number size of the semaphore, we deadlock forever. Will redo this later. * pubsub: make deprecated methods package-private (#1861) add back mistakenly deleted test Fixes #1828. * pubsub: rename newBuilder to defaultBuilder (#1873) Fixes #1853. * GAE (Flex_Java/Flex_Custom/Flex_Compat/Std_Java8), GCE, GKE testing app for gcj (#1859) Appengine tests for #1752 * Replace a constant of type Set with ImmutableSet (#1876) * Language v1beta2 Release (#1878) * Language v1beta2 Release * Language v1 update * Regenerating SPI: use setEndpoint (#1879) Reflects googleapis/gapic-generator#1172. Updates #1835. * remove last use of setPort/setServiceAddress (#1880) Fixes #1835. * new code snippet for push subscription + cleanup of deprecated snippets (#1875) * cleaning up PubSubExample adding snippet for creating a subscription with a push endpoint * updating start, end tags for snippets, adding async pull snippet tag * Release 0.12.0 * Updating version in README files. [ci skip] * Update version to 0.12.1-SNAPSHOT (#1886) * Update version of google-auth-java to 0.6.1 (#1888) * Update version of google-auth-java to 0.6.1 Latest version of google-auth-java contains a fix for auth token refresh failures. * Remove harcoded auth dependencies * Don't use `UrlFetchTransport` in App Engine Flex environment (#1893) * Don't use `UrlFetchTransport` in App Engine Flex environment #1492 * Add annotations to specify GCP launch stage (#1889) The Google Cloud Platform launch stage (https://cloud.google.com/terms/launch-stages) is a signifier of the level of access and support that can be expected of a particular feature. These annotations will be used to clearly demarcate features as being in a state other than General Availability to help set user expectations accordingly. * Add Identity Access Management (IAM) to the Storage API (#1812) Adds support for bucket-level IAM (currently in limited alpha). More information about IAM in Google Cloud Storage can be found at https://cloud.google.com/storage/docs/access-control/iam

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>

#1831)

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>

#1831)

…v1 to v2.54.0 (googleapis#1831)

…cies to v3.20.0 (#1831)

…cies to v3.20.0 (#1831)

#1831)

pubsub: acquire FlowController before releasing

02c617e

pongad requested a review from garrettjonesgoogle March 30, 2017 07:27

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Mar 30, 2017

garrettjonesgoogle reviewed Mar 30, 2017

View reviewed changes

Merge remote-tracking branch 'upstream/master' into sub-flake

9cf81c5

pr comment

28d348d

garrettjonesgoogle reviewed Apr 4, 2017

View reviewed changes

pr comment

d08b5f6

garrettjonesgoogle approved these changes Apr 5, 2017

View reviewed changes

pongad merged commit 3717ac6 into googleapis:master Apr 6, 2017

pongad deleted the sub-flake branch April 6, 2017 02:18

pongad added a commit that referenced this pull request Apr 6, 2017

Revert "pubsub: acquire FlowController before releasing (#1831)"

59ddf8c

This reverts commit 3717ac6.

pongad mentioned this pull request Apr 6, 2017

Revert "pubsub: acquire FlowController before releasing" #1872

Merged

meltsufin pushed a commit that referenced this pull request Dec 22, 2025

chore(main): release 3.23.1-SNAPSHOT (#1831)

e928268

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>

chingor13 pushed a commit that referenced this pull request Jan 22, 2026

chore: Update generation configuration at Tue Apr 29 02:28:51 UTC 2025 (

42188b9

#1831)

suztomo pushed a commit to suztomo/google-cloud-java that referenced this pull request Mar 11, 2026

chore(main): release 3.23.1-SNAPSHOT (googleapis#1831)

45d0618

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>

lqiu96 pushed a commit that referenced this pull request Mar 20, 2026

chore: Update generation configuration at Tue Apr 29 02:28:51 UTC 2025 (

156b0b1

#1831)

suztomo pushed a commit to suztomo/google-cloud-java that referenced this pull request Mar 23, 2026

deps: update dependency com.google.api.grpc:proto-google-cloud-trace-…

bda7038

…v1 to v2.54.0 (googleapis#1831)

meltsufin pushed a commit that referenced this pull request Apr 29, 2026

deps: update dependency com.google.cloud:google-cloud-shared-dependen…

d188af8

…cies to v3.20.0 (#1831)

meltsufin pushed a commit that referenced this pull request May 1, 2026

deps: update dependency com.google.cloud:google-cloud-shared-dependen…

38d81bf

…cies to v3.20.0 (#1831)

meltsufin pushed a commit that referenced this pull request May 2, 2026

chore: Update generation configuration at Wed Sep 18 02:22:45 UTC 2024 (

d75d4c8

#1831)

Conversation

pongad commented Mar 30, 2017

Uh oh!

coveralls commented Mar 30, 2017

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

pongad commented Apr 4, 2017

Uh oh!

coveralls commented Apr 4, 2017

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

coveralls commented Apr 5, 2017

Uh oh!

pongad commented Apr 5, 2017

Uh oh!

garrettjonesgoogle left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants