PR comments

pongad · pongad · commit 866195744dc5 · 2017-01-25T18:25:14.000+11:00
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Publisher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Publisher.java
@@ -23,6 +23,7 @@
 import com.google.api.gax.grpc.InstantiatingExecutorProvider;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
@@ -125,6 +126,7 @@ public static long getApiMaxRequestBytes() {
 
   private final BundlingSettings bundlingSettings;
   private final RetrySettings retrySettings;
+  private final LongRandom longRandom;
 
   private final FlowController.Settings flowControlSettings;
   private final boolean failOnFlowControlLimits;
@@ -151,6 +153,7 @@ private Publisher(Builder builder) throws IOException {
 
     this.bundlingSettings = builder.bundlingSettings;
     this.retrySettings = builder.retrySettings;
+    this.longRandom = builder.longRandom;
 
     flowControlSettings = builder.flowControlSettings;
     failOnFlowControlLimits = builder.failOnFlowControlLimits;
@@ -380,7 +383,8 @@ public void onSuccess(PublishResponse result) {
 
           @Override
           public void onFailure(Throwable t) {
-            long nextBackoffDelay = computeNextBackoffDelayMs(outstandingBundle, retrySettings);
+            long nextBackoffDelay =
+                computeNextBackoffDelayMs(outstandingBundle, retrySettings, longRandom);
 
             if (!isRetryable(t)
                 || System.currentTimeMillis() + nextBackoffDelay
@@ -494,14 +498,14 @@ private boolean hasBundlingBytes() {
   }
 
   private static long computeNextBackoffDelayMs(
-      OutstandingBundle outstandingBundle, RetrySettings retrySettings) {
+      OutstandingBundle outstandingBundle, RetrySettings retrySettings, LongRandom longRandom) {
     long delayMillis =
         Math.round(
             retrySettings.getInitialRetryDelay().getMillis()
                 * Math.pow(retrySettings.getRetryDelayMultiplier(), outstandingBundle.attempt - 1));
     delayMillis = Math.min(retrySettings.getMaxRetryDelay().getMillis(), delayMillis);
     outstandingBundle.attempt++;
-    return ThreadLocalRandom.current().nextLong(delayMillis / 2, delayMillis);
+    return longRandom.nextLong(0, delayMillis);
   }
 
   private boolean isRetryable(Throwable t) {
@@ -520,6 +524,10 @@ private boolean isRetryable(Throwable t) {
     }
   }
 
+  interface LongRandom {
+    long nextLong(long least, long bound);
+  }
+
   /** A builder of {@link Publisher}s. */
   public static final class Builder {
     static final Duration MIN_TOTAL_TIMEOUT = new Duration(10 * 1000); // 10 seconds
@@ -547,6 +555,13 @@ public static final class Builder {
             .setRpcTimeoutMultiplier(2)
             .setMaxRpcTimeout(DEFAULT_RPC_TIMEOUT)
             .build();
+    static final LongRandom DEFAULT_LONG_RANDOM =
+        new LongRandom() {
+          @Override
+          public long nextLong(long least, long bound) {
+            return ThreadLocalRandom.current().nextLong(least, bound);
+          }
+        };
 
     private static final int THREADS_PER_CPU = 5;
     static final ExecutorProvider DEFAULT_EXECUTOR_PROVIDER =
@@ -564,6 +579,7 @@ public static final class Builder {
     boolean failOnFlowControlLimits = false;
 
     RetrySettings retrySettings = DEFAULT_RETRY_SETTINGS;
+    LongRandom longRandom = DEFAULT_LONG_RANDOM;
 
     // Channels and credentials
     Optional<Credentials> userCredentials = Optional.absent();
@@ -659,6 +675,12 @@ public Builder setRetrySettings(RetrySettings retrySettings) {
       return this;
     }
 
+    @VisibleForTesting
+    Builder setLongRandom(LongRandom longRandom) {
+      this.longRandom = Preconditions.checkNotNull(longRandom);
+      return this;
+    }
+
     /** Gives the ability to set a custom executor to be used by the library. */
     public Builder setExecutorProvider(ExecutorProvider executorProvider) {
       this.executorProvider = Preconditions.checkNotNull(executorProvider);
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
@@ -458,8 +458,8 @@ public Builder setFlowControlSettings(FlowController.Settings flowControlSetting
     /**
      * Set acknowledgement expiration padding.
      *
-     * <p>This is the time accounted before a message expiration is to happen, so the
-     * {@link Subscriber} is able to send an ack extension beforehand.
+     * <p>This is the time accounted before a message expiration is to happen, so the {@link
+     * Subscriber} is able to send an ack extension beforehand.
      *
      * <p>This padding duration is configurable so you can account for network latency. A reasonable
      * number must be provided so messages don't expire because of network latency between when the
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/PublisherImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/PublisherImplTest.java
@@ -299,8 +299,8 @@ public void testPublishFailureRetries_exceededsRetryDuration() throws Exception
                     .build())
             .build(); // To demonstrate that reaching duration will trigger publish
 
-    // We use exponential backoff with randomness. 30 should be more than enough.
-    for (int i = 0; i < 30; ++i) {
+    // With exponential backoff, starting at 5ms we should have no more than 11 retries
+    for (int i = 0; i < 11; ++i) {
       testPublisherServiceImpl.addPublishError(new FakeException());
     }
     ListenableFuture<String> publishFuture1 = sendTestMessage(publisher, "A");
@@ -604,6 +604,13 @@ private Builder getTestPublisherBuilder() {
     return Publisher.Builder.newBuilder(TEST_TOPIC)
         .setCredentials(testCredentials)
         .setExecutorProvider(FixedExecutorProvider.create(fakeExecutor))
-        .setChannelBuilder(testChannelBuilder);
+        .setChannelBuilder(testChannelBuilder)
+        .setLongRandom(
+            new Publisher.LongRandom() {
+              @Override
+              public long nextLong(long least, long bound) {
+                return bound - 1;
+              }
+            });
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -458,8 +458,8 @@ public Builder setFlowControlSettings(FlowController.Settings flowControlSetting`
`458`	`458`	`/**`
`459`	`459`	`* Set acknowledgement expiration padding.`
`460`	`460`	`*`
`461`		`- * <p>This is the time accounted before a message expiration is to happen, so the`
`462`		`- * {@link Subscriber} is able to send an ack extension beforehand.`
	`461`	`+ * <p>This is the time accounted before a message expiration is to happen, so the {@link`
	`462`	`+ * Subscriber} is able to send an ack extension beforehand.`
`463`	`463`	`*`
`464`	`464`	`* <p>This padding duration is configurable so you can account for network latency. A reasonable`
`465`	`465`	`* number must be provided so messages don't expire because of network latency between when the`