---

sduskis · sduskis · commit af7e47043381 · 2019-05-15T11:22:30.000-04:00
yaml --- r: 35287 b: refs/heads/pubsub-ordering-keys c: cec8e94 h: refs/heads/master i: 35285: dcabe89 35283: 62f0d1b 35279: 3dd18e4
diff --git a/[refs] b/[refs]
@@ -155,7 +155,7 @@ refs/tags/v0.72.0: a7703f2593ba312c0b2dde6fdfd4f5c764bb55ac
 refs/tags/v0.73.0: 21241ea8be9439cc5764c4944cdce21d34ce4f9e
 refs/tags/v0.74.0: 9d1f733dbbf790de7b494418523b69c4a9a57638
 refs/heads/ignoretest: 23c412ae07af3d0ab1caa2d44d5bc5c0ccb8b31d
-refs/heads/pubsub-ordering-keys: 33ebb406282caeb1411ffa36145e51dd7b49bb50
+refs/heads/pubsub-ordering-keys: cec8e9417015eddde4e69768e304ba37f05a2e16
 refs/tags/v0.75.0: c3673089ae09a897c1b4cf7dfe167fe4f8ab32fb
 refs/tags/v0.76.0: 395b016826d3ddf9cb8b34919636df15a4dbd032
 refs/tags/v0.77.0: 28a85a77883ccf5d48f297fd0ef3b3dca6ce01f0
diff --git a/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -157,14 +157,13 @@ private Publisher(Builder builder) throws IOException {
     stubSettings
         .publishSettings()
         .setRetryableCodes(
-            EnumSet.of(
-                StatusCode.Code.ABORTED,
-                StatusCode.Code.CANCELLED,
-                StatusCode.Code.DEADLINE_EXCEEDED,
-                StatusCode.Code.INTERNAL,
-                StatusCode.Code.RESOURCE_EXHAUSTED,
-                StatusCode.Code.UNKNOWN,
-                StatusCode.Code.UNAVAILABLE))
+            StatusCode.Code.ABORTED,
+            StatusCode.Code.CANCELLED,
+            StatusCode.Code.DEADLINE_EXCEEDED,
+            StatusCode.Code.INTERNAL,
+            StatusCode.Code.RESOURCE_EXHAUSTED,
+            StatusCode.Code.UNKNOWN,
+            StatusCode.Code.UNAVAILABLE)
         .setRetrySettings(retrySettingsBuilder.build())
         .setBatchingSettings(BatchingSettings.newBuilder().setIsEnabled(false).build());
     this.publisherStub = GrpcPublisherStub.create(stubSettings.build());
@@ -241,7 +240,10 @@ public ApiFuture<String> publish(PubsubMessage message) {
     messagesWaiter.incrementPendingMessages(1);
 
     if (!batchesToSend.isEmpty()) {
+      // TODO: if this is not an ordering keys scenario, will this do anything?
       publishAllWithoutInflight();
+
+      // TODO: if this is an ordering keys scenario, is this safe without messagesBatchLock?
       for (final OutstandingBatch batch : batchesToSend) {
         logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
         executor.execute(
@@ -264,7 +266,9 @@ public void run() {
    *
    * @param key The key for which to resume publishing.
    */
-  public void resumePublish(String key) {
+  // TODO: make this public when Ordering keys is live
+  @BetaApi
+  void resumePublish(String key) {
     Preconditions.checkState(!shutdown.get(), "Cannot publish on a shut-down publisher.");
     sequentialExecutor.resumePublish(key);
   }
@@ -338,6 +342,8 @@ private void publishAllWithoutInflight() {
           // it's released, the order of publishing cannot be guaranteed if `publish()` is called
           // while this function is running. This locking mechanism needs to be improved if it
           // causes any performance degradation.
+
+          // TODO: Will this cause a performance problem for non-ordering keys scenarios?
           publishOutstandingBatch(batch.popOutstandingBatch());
           it.remove();
         }
@@ -389,20 +395,21 @@ public void onFailure(Throwable t) {
           }
         };
 
+    ApiFuture<PublishResponse> future;
     if (outstandingBatch.orderingKey == null || outstandingBatch.orderingKey.isEmpty()) {
-      ApiFutures.addCallback(publishCall(outstandingBatch), futureCallback, directExecutor());
+      future = publishCall(outstandingBatch);
     } else {
       // If ordering key is specified, publish the batch using the sequential executor.
-      ApiFuture<PublishResponse> future =
+      future =
           sequentialExecutor.submit(
               outstandingBatch.orderingKey,
               new Callable<ApiFuture<PublishResponse>>() {
                 public ApiFuture<PublishResponse> call() {
                   return publishCall(outstandingBatch);
                 }
               });
-      ApiFutures.addCallback(future, futureCallback, directExecutor());
     }
+    ApiFutures.addCallback(future, futureCallback, directExecutor());
   }
 
   private static final class OutstandingBatch {
@@ -673,6 +680,8 @@ public Builder setRetrySettings(RetrySettings retrySettings) {
     }
 
     /** Sets the message ordering option. */
+    // TODO: make this public when Ordering keys is live
+    @BetaApi
     Builder setEnableMessageOrdering(boolean enableMessageOrdering) {
       this.enableMessageOrdering = enableMessageOrdering;
       return this;
