Skip to content

Commit f52a817

Browse files
garrettjonesgooglegarrettjonesgoogle
authored
---
yaml --- r: 7905 b: refs/heads/tswast-patch-1 c: 8fedf78 h: refs/heads/master i: 7903: b530232
1 parent ce97398 commit f52a817

5 files changed

Lines changed: 54 additions & 21 deletions

File tree

‎[refs]‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,5 +57,5 @@ refs/tags/v0.18.0: 9d193c4c4b9d1c6f21515dd8e50836b9194ec9bb
5757
refs/tags/v0.19.0: e67b56e4d8dad5f9a7b38c9b2107c23c828f2ed5
5858
refs/tags/v0.20.0: 839f7fb7156535146aa1cb2c5aadd8d375d854e8
5959
refs/tags/v0.20.1: 370471f437f1f4f68a11e068df5cd6bf39edb1fa
60-
refs/heads/tswast-patch-1: f72746c6183378664a9326ee406d679926a7def4
60+
refs/heads/tswast-patch-1: 8fedf78a05efa542363e270fc4624eee63411dc5
6161
refs/heads/pubsub-streaming-pull: 19262b752ee874eb2ca3b950eb2aef44d5a5267b

‎branches/tswast-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616

1717
package com.google.cloud.pubsub.spi.v1;
1818

19-
import static com.google.cloud.pubsub.spi.v1.StatusUtil.isRetryable;
20-
2119
import com.google.api.gax.core.FlowController;
2220
import com.google.api.stats.Distribution;
2321
import com.google.auth.Credentials;
@@ -38,7 +36,6 @@
3836
import com.google.pubsub.v1.SubscriberGrpc.SubscriberFutureStub;
3937
import com.google.pubsub.v1.Subscription;
4038
import io.grpc.Channel;
41-
import io.grpc.StatusRuntimeException;
4239
import io.grpc.auth.MoreCallCredentials;
4340
import java.util.List;
4441
import java.util.concurrent.ScheduledExecutorService;
@@ -167,8 +164,12 @@ public void run() {
167164

168165
@Override
169166
public void onFailure(Throwable cause) {
170-
if (!(cause instanceof StatusRuntimeException)
171-
|| isRetryable(((StatusRuntimeException) cause).getStatus())) {
167+
if (!isAlive()) {
168+
// we don't care about subscription failures when we're no longer running.
169+
logger.log(Level.FINE, "pull failure after service no longer running", cause);
170+
return;
171+
}
172+
if (StatusUtil.isRetryable(cause)) {
172173
logger.log(Level.SEVERE, "Failed to pull messages (recoverable): ", cause);
173174
executor.schedule(
174175
new Runnable() {
@@ -183,14 +184,18 @@ public void run() {
183184
},
184185
backoff.getMillis(),
185186
TimeUnit.MILLISECONDS);
186-
return;
187+
} else {
188+
messageDispatcher.stop();
189+
notifyFailed(cause);
187190
}
188-
messageDispatcher.stop();
189-
notifyFailed(cause);
190191
}
191192
});
192193
}
193194

195+
private boolean isAlive() {
196+
return state() == State.RUNNING || state() == State.STARTING;
197+
}
198+
194199
@Override
195200
public void sendAckOperations(
196201
List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions) {

‎branches/tswast-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StatusUtil.java‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,21 +17,31 @@
1717
package com.google.cloud.pubsub.spi.v1;
1818

1919
import io.grpc.Status;
20+
import io.grpc.StatusRuntimeException;
2021

2122
/** Utilities for handling gRPC {@link Status}. */
2223
final class StatusUtil {
2324
private StatusUtil() {
24-
// Static class, not instatiable.
25+
// Static class, not instantiable.
2526
}
2627

27-
public static boolean isRetryable(Status status) {
28-
switch (status.getCode()) {
28+
public static boolean isRetryable(Throwable error) {
29+
if (!(error instanceof StatusRuntimeException)) {
30+
return true;
31+
}
32+
StatusRuntimeException statusRuntimeException = (StatusRuntimeException) error;
33+
switch (statusRuntimeException.getStatus().getCode()) {
2934
case DEADLINE_EXCEEDED:
3035
case INTERNAL:
3136
case CANCELLED:
3237
case RESOURCE_EXHAUSTED:
33-
case UNAVAILABLE:
3438
return true;
39+
case UNAVAILABLE:
40+
if (statusRuntimeException.getMessage().contains("Server shutdownNow invoked")) {
41+
return false;
42+
} else {
43+
return true;
44+
}
3545
default:
3646
return false;
3747
}

‎branches/tswast-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StreamingSubscriberConnection.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616

1717
package com.google.cloud.pubsub.spi.v1;
1818

19-
import static com.google.cloud.pubsub.spi.v1.StatusUtil.isRetryable;
20-
2119
import com.google.api.gax.core.FlowController;
2220
import com.google.api.stats.Distribution;
2321
import com.google.auth.Credentials;
@@ -179,9 +177,13 @@ public void onSuccess(@Nullable Void result) {
179177
}
180178

181179
@Override
182-
public void onFailure(Throwable t) {
183-
Status errorStatus = Status.fromThrowable(t);
184-
if (isRetryable(errorStatus) && isAlive()) {
180+
public void onFailure(Throwable cause) {
181+
if (!isAlive()) {
182+
// we don't care about subscription failures when we're no longer running.
183+
logger.log(Level.FINE, "pull failure after service no longer running", cause);
184+
return;
185+
}
186+
if (StatusUtil.isRetryable(cause)) {
185187
long backoffMillis = channelReconnectBackoff.getMillis();
186188
channelReconnectBackoff = channelReconnectBackoff.plus(backoffMillis);
187189
executor.schedule(
@@ -194,9 +196,7 @@ public void run() {
194196
backoffMillis,
195197
TimeUnit.MILLISECONDS);
196198
} else {
197-
if (isAlive()) {
198-
notifyFailed(t);
199-
}
199+
notifyFailed(cause);
200200
}
201201
}
202202
},

‎branches/tswast-patch-1/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/StatusUtilTest.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
package com.google.cloud.pubsub.spi.v1;
2+
3+
import com.google.common.truth.Truth;
4+
import io.grpc.Status;
5+
import io.grpc.StatusRuntimeException;
6+
import org.junit.Test;
7+
8+
public class StatusUtilTest {
9+
10+
@Test
11+
public void testIsRetryable() {
12+
Truth.assertThat(StatusUtil.isRetryable(new StatusRuntimeException(Status.UNAVAILABLE)))
13+
.isTrue();
14+
Truth.assertThat(StatusUtil.isRetryable(new StatusRuntimeException(
15+
Status.UNAVAILABLE.withDescription("Server shutdownNow invoked"))))
16+
.isFalse();
17+
}
18+
}

0 commit comments

Comments
 (0)