---

pongad · garrettjonesgoogle · commit 180f1a60390b · 2017-02-15T08:39:06.000-08:00
yaml --- r: 5389 b: refs/heads/master c: 68ce84d h: refs/heads/master i: 5387: a5d011d
diff --git a/[refs] b/[refs]
@@ -1,5 +1,5 @@
 ---
-refs/heads/master: 093631b41ba0b6e7702c4d89991e5ac31b291caa
+refs/heads/master: 68ce84d313b7745b1dc37be300522ea1c051063b
 refs/heads/travis: dae77e558b884bc1b165155482d76c8e40b0fca4
 refs/heads/gh-pages: 229631582f8957646f81e92ae5a326504f48ee5b
 refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444
diff --git a/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java b/trunk/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/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java b/trunk/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/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java b/trunk/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/trunk/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeSubscriberServiceImpl.java b/trunk/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/trunk/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java b/trunk/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`