Skip to content

Commit 315e58a

Browse files
pongadpongad
committed
pubsub: enable streaming pull
Moving this to a branch. If nothing else, we should perf test this before declaring it release-ready. This commit re-enables streaming pull. Unlike previous implementation, it does not fall back to polling if streaming is unavailable, since the streaming pull endpoint should be working by the time this is released. This commit fixes a bug deadline modification code. Previously we record an Instant we receive a message, then call Instant::getNano to get the time in nanoseconds. This is incorrect since getNano returns the nanosecond from the beginning of that second, not since an epoch. The fix is to simply save the time since epoch instead of Instant. This commit also slightly changes how errors are logged, so that errors that occur after the service is being shut down don't spam the console.
1 parent 2a92207 commit 315e58a

4 files changed

Lines changed: 121 additions & 28 deletions

File tree

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -173,13 +173,13 @@ private class AckHandler implements FutureCallback<AckReply> {
173173
private final String ackId;
174174
private final int outstandingBytes;
175175
private final AtomicBoolean acked;
176-
private final Instant receivedTime;
176+
private final long receivedTimeMillis;
177177

178178
AckHandler(String ackId, int outstandingBytes) {
179179
this.ackId = ackId;
180180
this.outstandingBytes = outstandingBytes;
181181
acked = new AtomicBoolean(false);
182-
receivedTime = Instant.ofEpochMilli(clock.millisTime());
182+
receivedTimeMillis = clock.millisTime();
183183
}
184184

185185
@Override
@@ -207,7 +207,6 @@ public void onSuccess(AckReply reply) {
207207
pendingAcks.add(ackId);
208208
}
209209
// Record the latency rounded to the next closest integer.
210-
long receivedTimeMillis = TimeUnit.NANOSECONDS.toMillis(receivedTime.getNano());
211210
ackLatencyDistribution.record(
212211
Ints.saturatedCast(
213212
(long) Math.ceil((clock.millisTime() - receivedTimeMillis) / 1000D)));

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

Lines changed: 10 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,9 @@
2929
import com.google.common.util.concurrent.SettableFuture;
3030
import com.google.pubsub.v1.StreamingPullRequest;
3131
import com.google.pubsub.v1.StreamingPullResponse;
32-
import com.google.pubsub.v1.SubscriberGrpc;
33-
import io.grpc.CallOptions;
34-
import io.grpc.Channel;
32+
import com.google.pubsub.v1.SubscriberGrpc.SubscriberStub;
3533
import io.grpc.Status;
3634
import io.grpc.stub.ClientCallStreamObserver;
37-
import io.grpc.stub.ClientCalls;
3835
import io.grpc.stub.ClientResponseObserver;
3936
import java.util.ArrayList;
4037
import java.util.List;
@@ -55,7 +52,7 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
5552

5653
private Duration channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
5754

58-
private final Channel channel;
55+
private final SubscriberStub asyncStub;
5956

6057
private final String subscription;
6158
private final ScheduledExecutorService executor;
@@ -69,14 +66,14 @@ public StreamingSubscriberConnection(
6966
Duration maxAckExtensionPeriod,
7067
int streamAckDeadlineSeconds,
7168
Distribution ackLatencyDistribution,
72-
Channel channel,
69+
SubscriberStub asyncStub,
7370
FlowController flowController,
7471
ScheduledExecutorService executor,
7572
@Nullable ScheduledExecutorService alarmsExecutor,
7673
ApiClock clock) {
7774
this.subscription = subscription;
7875
this.executor = executor;
79-
this.channel = channel;
76+
this.asyncStub = asyncStub;
8077
this.messageDispatcher =
8178
new MessageDispatcher(
8279
receiver,
@@ -101,8 +98,8 @@ protected void doStart() {
10198
@Override
10299
protected void doStop() {
103100
messageDispatcher.stop();
104-
notifyStopped();
105101
requestObserver.onError(Status.CANCELLED.asException());
102+
notifyStopped();
106103
}
107104

108105
private class StreamingPullResponseObserver
@@ -137,7 +134,6 @@ public void run() {
137134

138135
@Override
139136
public void onError(Throwable t) {
140-
logger.log(Level.WARNING, "Terminated streaming with exception", t);
141137
errorFuture.setException(t);
142138
}
143139

@@ -154,9 +150,7 @@ private void initialize() {
154150
new StreamingPullResponseObserver(errorFuture);
155151
final ClientCallStreamObserver<StreamingPullRequest> requestObserver =
156152
(ClientCallStreamObserver<StreamingPullRequest>)
157-
(ClientCalls.asyncBidiStreamingCall(
158-
channel.newCall(SubscriberGrpc.METHOD_STREAMING_PULL, CallOptions.DEFAULT),
159-
responseObserver));
153+
(asyncStub.streamingPull(responseObserver));
160154
logger.log(
161155
Level.FINER,
162156
"Initializing stream to subscription {0} with deadline {1}",
@@ -173,6 +167,9 @@ private void initialize() {
173167
new FutureCallback<Void>() {
174168
@Override
175169
public void onSuccess(@Nullable Void result) {
170+
if (!isAlive()) {
171+
return;
172+
}
176173
channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
177174
// The stream was closed. And any case we want to reopen it to continue receiving
178175
// messages.
@@ -186,6 +183,7 @@ public void onFailure(Throwable cause) {
186183
logger.log(Level.FINE, "pull failure after service no longer running", cause);
187184
return;
188185
}
186+
logger.log(Level.WARNING, "Terminated streaming with exception", cause);
189187
if (StatusUtil.isRetryable(cause)) {
190188
long backoffMillis = channelReconnectBackoff.toMillis();
191189
channelReconnectBackoff = channelReconnectBackoff.plusMillis(backoffMillis);

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java‎

Lines changed: 105 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -38,9 +38,11 @@
3838
import com.google.pubsub.v1.GetSubscriptionRequest;
3939
import com.google.pubsub.v1.SubscriberGrpc;
4040
import com.google.pubsub.v1.SubscriberGrpc.SubscriberFutureStub;
41+
import com.google.pubsub.v1.SubscriberGrpc.SubscriberStub;
4142
import com.google.pubsub.v1.Subscription;
4243
import com.google.pubsub.v1.SubscriptionName;
4344
import io.grpc.CallCredentials;
45+
import io.grpc.Channel;
4446
import io.grpc.ManagedChannel;
4547
import io.grpc.auth.MoreCallCredentials;
4648
import java.io.IOException;
@@ -257,22 +259,25 @@ public void close() {
257259
// same executor, it will deadlock: the thread will be stuck waiting for connections
258260
// to start but cannot start the connections.
259261
// For this reason, we spawn a dedicated thread. Starting subscriber should be rare.
260-
new Thread(new Runnable() {
261-
@Override
262-
public void run() {
263-
try {
264-
startPollingConnections();
265-
notifyStarted();
266-
} catch (Throwable t) {
267-
notifyFailed(t);
268-
}
269-
}
270-
}).start();
262+
new Thread(
263+
new Runnable() {
264+
@Override
265+
public void run() {
266+
try {
267+
// startPollingConnections();
268+
startStreamingConnections();
269+
notifyStarted();
270+
} catch (Throwable t) {
271+
notifyFailed(t);
272+
}
273+
}
274+
})
275+
.start();
271276
}
272277

273278
@Override
274279
protected void doStop() {
275-
// stopAllStreamingConnections();
280+
stopAllStreamingConnections();
276281
stopAllPollingConnections();
277282
try {
278283
for (AutoCloseable closeable : closeables) {
@@ -284,6 +289,94 @@ protected void doStop() {
284289
}
285290
}
286291

292+
private void startStreamingConnections() throws IOException {
293+
synchronized (streamingSubscriberConnections) {
294+
Credentials credentials = credentialsProvider.getCredentials();
295+
CallCredentials callCredentials =
296+
credentials == null ? null : MoreCallCredentials.from(credentials);
297+
298+
for (Channel channel : channels) {
299+
SubscriberStub stub = SubscriberGrpc.newStub(channel);
300+
if (callCredentials != null) {
301+
stub = stub.withCallCredentials(callCredentials);
302+
}
303+
streamingSubscriberConnections.add(
304+
new StreamingSubscriberConnection(
305+
cachedSubscriptionNameString,
306+
receiver,
307+
ackExpirationPadding,
308+
maxAckExtensionPeriod,
309+
streamAckDeadlineSeconds,
310+
ackLatencyDistribution,
311+
stub,
312+
flowController,
313+
executor,
314+
alarmsExecutor,
315+
clock));
316+
}
317+
startConnections(
318+
streamingSubscriberConnections,
319+
new Listener() {
320+
@Override
321+
public void failed(State from, Throwable failure) {
322+
// If a connection failed is because of a fatal error, we should fail the
323+
// whole subscriber.
324+
stopAllStreamingConnections();
325+
try {
326+
notifyFailed(failure);
327+
} catch (IllegalStateException e) {
328+
if (isRunning()) {
329+
throw e;
330+
}
331+
// It could happen that we are shutting down while some channels fail.
332+
}
333+
}
334+
});
335+
}
336+
337+
ackDeadlineUpdater =
338+
executor.scheduleAtFixedRate(
339+
new Runnable() {
340+
@Override
341+
public void run() {
342+
// It is guaranteed this will be <= MAX_ACK_DEADLINE_SECONDS, the max of the API.
343+
long ackLatency =
344+
ackLatencyDistribution.getNthPercentile(PERCENTILE_FOR_ACK_DEADLINE_UPDATES);
345+
if (ackLatency > 0) {
346+
long ackExpirationPaddingMillis = ackExpirationPadding.toMillis();
347+
int possibleStreamAckDeadlineSeconds =
348+
Math.max(
349+
MIN_ACK_DEADLINE_SECONDS,
350+
Ints.saturatedCast(
351+
Math.max(
352+
ackLatency,
353+
TimeUnit.MILLISECONDS.toSeconds(ackExpirationPaddingMillis))));
354+
if (streamAckDeadlineSeconds != possibleStreamAckDeadlineSeconds) {
355+
streamAckDeadlineSeconds = possibleStreamAckDeadlineSeconds;
356+
logger.log(
357+
Level.FINER,
358+
"Updating stream deadline to {0} seconds.",
359+
streamAckDeadlineSeconds);
360+
for (StreamingSubscriberConnection subscriberConnection :
361+
streamingSubscriberConnections) {
362+
subscriberConnection.updateStreamAckDeadline(streamAckDeadlineSeconds);
363+
}
364+
}
365+
}
366+
}
367+
},
368+
ACK_DEADLINE_UPDATE_PERIOD.toMillis(),
369+
ACK_DEADLINE_UPDATE_PERIOD.toMillis(),
370+
TimeUnit.MILLISECONDS);
371+
}
372+
373+
private void stopAllStreamingConnections() {
374+
stopConnections(streamingSubscriberConnections);
375+
if (ackDeadlineUpdater != null) {
376+
ackDeadlineUpdater.cancel(true);
377+
}
378+
}
379+
287380
// Starts polling connections. Blocks until all connections declare themselves running.
288381
private void startPollingConnections() throws IOException {
289382
synchronized (pollingSubscriberConnections) {

‎google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ public class SubscriberTest {
8080

8181
@Parameters
8282
public static Collection<Object[]> data() {
83-
return Arrays.asList(new Object[][] {{false}});
83+
return Arrays.asList(new Object[][] {{true}});
8484
}
8585

8686
static class TestReceiver implements MessageReceiver {
@@ -205,6 +205,9 @@ public void testAckSingleMessage() throws Exception {
205205

206206
@Test
207207
public void testGetSubscriptionOnce() throws Exception {
208+
if (isStreamingTest) {
209+
return;
210+
}
208211
Subscriber subscriber = startSubscriber(getTestSubscriberBuilder(testReceiver));
209212

210213
sendMessages(ImmutableList.of("A"));

0 commit comments

Comments
 (0)