Skip to content

pubsub: fix race condition in streaming connection#2416

Merged
pongad merged 8 commits intogoogleapis:masterfrom
pongad:stream-race
Oct 4, 2017
Merged

pubsub: fix race condition in streaming connection#2416
pongad merged 8 commits intogoogleapis:masterfrom
pongad:stream-race

Conversation

@pongad
Copy link
Copy Markdown
Contributor

@pongad pongad commented Sep 8, 2017

No description provided.

@googlebot googlebot added the cla: yes This human has signed the Contributor License Agreement. label Sep 8, 2017
@pongad
Copy link
Copy Markdown
Contributor Author

pongad commented Sep 8, 2017

@davidtorres @mdietz94

mdietz94
mdietz94 suggested changes Sep 9, 2017
View reviewed changes
Copy link
Copy Markdown

@mdietz94 mdietz94 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would replace ReentrantLock with ReadWriteLock. I think it will simplify the concurrency model + we can allow almost all operations here to happen concurrently, except for recreating the stream so that we only call Ack on the new stream.

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
private final AtomicLong channelReconnectBackoffMillis =
new AtomicLong(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());

private final Lock lock = new ReentrantLock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java

lock.lock();
try {
requestObserver.onError(Status.CANCELLED.asException());

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Outdated
StreamingSubscriberConnection.this.requestObserver = requestObserver;
requestObserver.disableAutoInboundFlowControl();
thisRequestObserver = requestObserver;
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
requestObserver.request(1);
lock.lock();
try {
thisRequestObserver.request(1);

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
partitionAckOperations(acksToSend, ackDeadlineExtensions, MAX_PER_REQUEST_CHANGES);
for (StreamingPullRequest request : requests) {
requestObserver.onNext(request);
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
StreamingPullRequest.newBuilder()
.setStreamAckDeadlineSeconds(newAckDeadlineSeconds)
.build());
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

mdietz94
mdietz94 suggested changes Sep 11, 2017
View reviewed changes
Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
requestObserver.request(1);
lock.lock();
try {
thisRequestObserver.request(1);

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
partitionAckOperations(acksToSend, ackDeadlineExtensions, MAX_PER_REQUEST_CHANGES);
for (StreamingPullRequest request : requests) {
requestObserver.onNext(request);
lock.lock();

This comment was marked as spam.

Sign in to view

Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
StreamingPullRequest.newBuilder()
.setStreamAckDeadlineSeconds(newAckDeadlineSeconds)
.build());
lock.lock();

This comment was marked as spam.

Sign in to view

@pongad
Copy link
Copy Markdown
Contributor Author

pongad commented Sep 13, 2017

@mdietz94 PTAL

mdietz94
mdietz94 reviewed Sep 14, 2017
View reviewed changes
Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Outdated
} finally {
lock.unlock();
}
thisRequestObserver.request(1);

This comment was marked as spam.

Sign in to view

mdietz94
mdietz94 suggested changes Sep 14, 2017
View reviewed changes
Copy link
Copy Markdown

@mdietz94 mdietz94 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Still require a reader lock around request to avoid concurrent access with onNext/onError

@pongad
Copy link
Copy Markdown
Contributor Author

pongad commented Sep 14, 2017

Thinking about this more, I think I found more issues with streams. I'll follow up with you and David on email.

@pongad pongad mentioned this pull request Sep 28, 2017
Closed
@pongad
Copy link
Copy Markdown
Contributor Author

pongad commented Sep 28, 2017

@mdietz94 @davidcavazos Please let me know if this looks OK to you. The error handling is still obviously wrong. However, I think this is a big enough improvement that we can submit this as an intermediate state.

  • At least we don't have the race condition anymore.
  • If onNext throws, it might stop us from scheduling the next deadline extension. After this change, at least it can't do crazy things like that anymore. It will make us loose some acks and stuff like that, but that's still a lot better.

@davidcavazos
Copy link
Copy Markdown

davidcavazos commented Sep 28, 2017

Did you mean @davidtorres ?

@pongad
Copy link
Copy Markdown
Contributor Author

pongad commented Sep 29, 2017

Yes I did! Sorry for the spam!

@pongad
Copy link
Copy Markdown
Contributor Author

pongad commented Sep 29, 2017

@davidtorres

mdietz94
mdietz94 suggested changes Oct 4, 2017
View reviewed changes
Comment thread google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
// If errorFuture is done, the stream has either failed or hung up,
// and we don't need to request.
if (isAlive() && !errorFuture.isDone()) {
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

@pongad pongad merged commit 97d50c2 into googleapis:master Oct 4, 2017
@pongad pongad deleted the stream-race branch October 4, 2017 02:55
chingor13 pushed a commit that referenced this pull request Feb 20, 2026
@renovate-bot @gcf-owl-bot
chore(deps): update dependency com.google.cloud:libraries-bom to v26.…
0b931f0 
…33.0 (#2416)

* chore(deps): update dependency com.google.cloud:libraries-bom to v26.33.0

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
meltsufin pushed a commit that referenced this pull request Apr 29, 2026
@release-please @cloud-java-bot
chore(main): release 1.139.3 (#2416)
58d9cb2 
* chore(main): release 1.139.3

* chore: generate libraries at Tue May  6 15:01:24 UTC 2025

---------

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>
Co-authored-by: cloud-java-bot <[email protected]>
meltsufin pushed a commit that referenced this pull request May 1, 2026
@release-please @cloud-java-bot
chore(main): release 1.139.3 (#2416)
b2a5925 
* chore(main): release 1.139.3

* chore: generate libraries at Tue May  6 15:01:24 UTC 2025

---------

Co-authored-by: release-please[bot] <55107282+release-please[bot]@users.noreply.github.com>
Co-authored-by: cloud-java-bot <[email protected]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@garrettjonesgoogle garrettjonesgoogle Awaiting requested review from garrettjonesgoogle

1 more reviewer

@mdietz94 mdietz94 mdietz94 requested changes

Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

cla: yes This human has signed the Contributor License Agreement.

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

4 participants

@pongad @davidcavazos @mdietz94 @googlebot