Skip to content
Permalink
Browse files
fix: properly shutdown subscriber stub on permanent streaming pull fa…
…ilure (#539)

* fix: stop the subscriber stub on streaming pull failure
  • Loading branch information
hannahrogers-google committed Mar 3, 2021
1 parent 373aee2 commit adbcc0c6777e35eae24b538e6c48f9ef7485a786
@@ -149,8 +149,7 @@ protected void doStart() {

@Override
protected void doStop() {
messageDispatcher.stop();
ackOperationsWaiter.waitComplete();
runShutdown();

lock.lock();
try {
@@ -161,6 +160,11 @@ protected void doStop() {
}
}

private void runShutdown() {
messageDispatcher.stop();
ackOperationsWaiter.waitComplete();
}

private class StreamingPullResponseObserver implements ResponseObserver<StreamingPullResponse> {

final SettableApiFuture<Void> errorFuture;
@@ -282,6 +286,7 @@ public void onFailure(Throwable cause) {
ApiExceptionFactory.createException(
cause, GrpcStatusCode.of(Status.fromThrowable(cause).getCode()), false);
logger.log(Level.SEVERE, "terminated streaming with exception", gaxException);
runShutdown();
notifyFailed(gaxException);
return;
}
@@ -308,9 +308,7 @@ protected void doStop() {
public void run() {
try {
// stop connection is no-op if connections haven't been started.
stopAllStreamingConnections();
shutdownBackgroundResources();
subStub.shutdownNow();
runShutdown();
notifyStopped();
} catch (Exception e) {
notifyFailed(e);
@@ -320,6 +318,12 @@ public void run() {
.start();
}

private void runShutdown() {
stopAllStreamingConnections();
shutdownBackgroundResources();
subStub.shutdownNow();
}

private void startStreamingConnections() {
synchronized (streamingSubscriberConnections) {
for (int i = 0; i < numPullers; i++) {
@@ -352,8 +356,7 @@ private void startStreamingConnections() {
public void failed(State from, Throwable failure) {
// If a connection failed is because of a fatal error, we should fail the
// whole subscriber.
stopAllStreamingConnections();
shutdownBackgroundResources();
runShutdown();
try {
notifyFailed(failure);
} catch (IllegalStateException e) {

0 comments on commit adbcc0c

Please sign in to comment.