pubsub: start with polling by default (#1625)

pongad · garrettjonesgoogle · commit 68ce84d313b7 · 2017-02-15T08:39:06.000-08:00
* pubsub: start with polling by default

Previously, Subscribers start with streaming pull and fall back
to polling pull if streaming is unavailable.
To prevent fallback or streaming from causing any problem in production,
this commit starts Subscribers with polling pull directly.
Streaming pull will be re-enabled once the endpoint is working.

Since the ultimate fate of fallback is up in the air,
I chose to comment out the start-with-streaming code instead of
deleting.

This change also smoked out a logic bug that causes a race condition.
A message can be "nacked" either by calling AckReplyConsumer::accept
with an explicit NACK or a throwable signally an error.
The explicit NACK case works properly.

In case of a throwable, MessageDispatcher did not set the "acked" flag.
(Acks and nacks share most of the same code path;
they might as well use the same flag.)
This causes two things to happen concurrently.

1. Since the message is being nacked, it is added to the nack list
to be reported to the pubsub service.

2. Pubsub client automatically extends deadlines of the messages
the client's user is processing. Since acked flag is not set,
the client also tries to extend the message's deadline.

If (1) happens first, the test code sees that the message is being
nacked and the test passes even though the client will later
incorrectly extend the message's deadline.

If (2) happens first, the test code sees the incorrect deadline
extension and fails.

The fix is simple: set the acked flag.

* pubsub: fix testModifyAckDeadline flaking, maybe

Previously MessageDispatcher starts off with deadline duration of 0.
Since we need to extend deadline before the time is up,
the client extends deadline a little before the deadline.
This translates to scheduling deadling extension "in the past".

In turn, this causes tasks to be run -- and new tasks scheduled --
without explicit calls to advance the time on the fake clock.
So, when the test code actually advance the fake clock,
it runs more tasks than it wants to.

This PR conservatively sets the initial deadline duration to 10 seconds.
This is already the default in streaming version.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java
@@ -169,6 +169,7 @@ public void onFailure(Throwable t) {
           Level.WARNING,
           "MessageReceiver failed to processes ack ID: " + ackId + ", the message will be nacked.",
           t);
+      acked.getAndSet(true);
       synchronized (pendingNacks) {
         pendingNacks.add(ackId);
       }
@@ -494,7 +495,6 @@ private void processOutstandingAckOperations(
         modifyAckDeadlinesToSend.add(nacksToSend);
       }
     }
-
     ackProcessor.sendAckOperations(acksToSend, modifyAckDeadlinesToSend);
   }
 }
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java
@@ -90,6 +90,7 @@ public PollingSubscriberConnection(
             flowController,
             executor,
             clock);
+    messageDispatcher.setMessageDeadlineSeconds(Subscriber.MIN_ACK_DEADLINE_SECONDS);
   }
 
   @Override
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java
@@ -113,7 +113,7 @@ public class Subscriber {
       20 * 1024 * 1024; // 20MB API maximum message size.
   private static final int INITIAL_ACK_DEADLINE_SECONDS = 10;
   private static final int MAX_ACK_DEADLINE_SECONDS = 600;
-  private static final int MIN_ACK_DEADLINE_SECONDS = 10;
+  static final int MIN_ACK_DEADLINE_SECONDS = 10;
   private static final Duration ACK_DEADLINE_UPDATE_PERIOD = Duration.standardMinutes(1);
   private static final double PERCENTILE_FOR_ACK_DEADLINE_UPDATES = 99.9;
 
@@ -333,7 +333,9 @@ public void close() throws IOException {
     @Override
     protected void doStart() {
       logger.log(Level.INFO, "Starting subscriber group.");
-      startStreamingConnections();
+      // Streaming pull is not enabled on the service yet.
+      // startStreamingConnections();
+      startPollingConnections();
       notifyStarted();
     }
 
@@ -422,7 +424,9 @@ public void run() {
 
     private void stopAllStreamingConnections() {
       stopConnections(streamingSubscriberConnections);
-      ackDeadlineUpdater.cancel(true);
+      if (ackDeadlineUpdater != null) {
+        ackDeadlineUpdater.cancel(true);
+      }
     }
 
     private void startPollingConnections() {
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeSubscriberServiceImpl.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeSubscriberServiceImpl.java
@@ -49,7 +49,7 @@
 class FakeSubscriberServiceImpl extends SubscriberImplBase {
   private final AtomicBoolean subscriptionInitialized = new AtomicBoolean(false);
   private String subscription = "";
-  private final AtomicInteger messageAckDeadline = new AtomicInteger();
+  private final AtomicInteger messageAckDeadline = new AtomicInteger(Subscriber.MIN_ACK_DEADLINE_SECONDS);
   private final List<Stream> openedStreams = new ArrayList<>();
   private final List<Stream> closedStreams = new ArrayList<>();
   private final List<String> acks = new ArrayList<>();
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java
@@ -68,7 +68,7 @@ public class SubscriberImplTest {
 
   @Parameters
   public static Collection<Object[]> data() {
-    return Arrays.asList(new Object[][] {{true}, {false}});
+    return Arrays.asList(new Object[][] {{false}});
   }
 
   private final boolean isStreamingTest;
@@ -436,10 +436,10 @@ public void testFailedChannel_fatalError_subscriberFails() throws Exception {
   private Subscriber startSubscriber(Builder testSubscriberBuilder) throws Exception {
     Subscriber subscriber = testSubscriberBuilder.build();
     subscriber.startAsync().awaitRunning();
-    if (!isStreamingTest) {
-      // Shutdown streaming
-      fakeSubscriberServiceImpl.sendError(new StatusException(Status.UNIMPLEMENTED));
-    }
+    // if (!isStreamingTest) {
+    //   // Shutdown streaming
+    //   fakeSubscriberServiceImpl.sendError(new StatusException(Status.UNIMPLEMENTED));
+    // }
     return subscriber;
   }
 

Original file line number	Diff line number	Diff line change
`@@ -169,6 +169,7 @@ public void onFailure(Throwable t) {`
`169`	`169`	`Level.WARNING,`
`170`	`170`	`"MessageReceiver failed to processes ack ID: " + ackId + ", the message will be nacked.",`
`171`	`171`	`t);`
	`172`	`+ acked.getAndSet(true);`
`172`	`173`	`synchronized (pendingNacks) {`
`173`	`174`	`pendingNacks.add(ackId);`
`174`	`175`	`}`
`@@ -494,7 +495,6 @@ private void processOutstandingAckOperations(`
`494`	`495`	`modifyAckDeadlinesToSend.add(nacksToSend);`
`495`	`496`	`}`
`496`	`497`	`}`
`497`		`-`
`498`	`498`	`ackProcessor.sendAckOperations(acksToSend, modifyAckDeadlinesToSend);`
`499`	`499`	`}`
`500`	`500`	`}`
Original file line number	Diff line number	Diff line change
`@@ -90,6 +90,7 @@ public PollingSubscriberConnection(`
`90`	`90`	`flowController,`
`91`	`91`	`executor,`
`92`	`92`	`clock);`
	`93`	`+ messageDispatcher.setMessageDeadlineSeconds(Subscriber.MIN_ACK_DEADLINE_SECONDS);`
`93`	`94`	`}`
`94`	`95`
`95`	`96`	`@Override`