---

sduskis · sduskis · commit 223267a103d2 · 2019-04-22T18:22:41.000-04:00
yaml --- r: 27255 b: refs/heads/pubsub-ordering-keys c: 97ba17c h: refs/heads/master i: 27253: eefc969 27251: 33d39e3 27247: 39a72d2
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: dd8a5cef1c2a6038210b87c6e5a581000cb14409
+refs/heads/pubsub-ordering-keys: 97ba17c86f930ff5226d2da4fffa26a183b03d65
 "refs/heads/update_mvn_badge": ae2d773814db0f71197ccf5a8612ee1d8056f8de
 refs/tags/v0.75.0: c3673089ae09a897c1b4cf7dfe167fe4f8ab32fb
 refs/tags/v0.76.0: 395b016826d3ddf9cb8b34919636df15a4dbd032
diff --git a/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/SequentialExecutorService.java b/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/SequentialExecutorService.java
@@ -30,8 +30,6 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.Executor;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
 interface CancellableRunnable extends Runnable {
   void cancel(Throwable e);
@@ -52,19 +50,19 @@ private SequentialExecutorService() {}
    * tasks with the same key sequentially. Tasks with the same key will be run only when its
    * predecessor has been completed while tasks with different keys can be run in parallel.
    */
-  private abstract static class SequentialExecutor {
+  private abstract static class SequentialExecutor<R extends Runnable> {
     // Maps keys to tasks.
-    protected final Map<String, Deque<Runnable>> tasksByKey;
+    protected final Map<String, Deque<R>> tasksByKey;
     protected final Executor executor;
 
     private SequentialExecutor(Executor executor) {
       this.executor = executor;
       this.tasksByKey = new HashMap<>();
     }
 
-    protected void execute(final String key, Runnable task) {
+    protected void execute(final String key, R task) {
       synchronized (tasksByKey) {
-        Deque<Runnable> newTasks = tasksByKey.get(key);
+        Deque<R> newTasks = tasksByKey.get(key);
         // If this key is already being handled, add it to the queue and return.
         if (newTasks != null) {
           newTasks.add(task);
@@ -84,7 +82,7 @@ protected void callNextTaskAsync(final String key) {
           new Runnable() {
             @Override
             public void run() {
-              Deque<Runnable> tasks;
+              Deque<R> tasks;
               synchronized (tasksByKey) {
                 tasks = tasksByKey.get(key);
                 if (tasks != null && tasks.isEmpty()) {
@@ -94,7 +92,7 @@ public void run() {
               }
               if (tasks != null) {
                 // TODO(kimkyung-goog): Check if there is a race when task list becomes empty.
-                Runnable task = tasks.poll();
+                R task = tasks.poll();
                 if (task != null) {
                   task.run();
                 }
@@ -105,7 +103,7 @@ public void run() {
   }
 
   @BetaApi
-  static class AutoExecutor extends SequentialExecutor {
+  static class AutoExecutor extends SequentialExecutor<Runnable> {
     AutoExecutor(Executor executor) {
       super(executor);
     }
@@ -126,10 +124,7 @@ void submit(final String key, final Runnable task) {
    * fails, other tasks with the same key that have not been executed will be cancelled.
    */
   @BetaApi
-  static class CallbackExecutor extends SequentialExecutor {
-    private static final Logger logger =
-        Logger.getLogger(SequentialExecutorService.SequentialExecutor.class.getName());
-
+  static class CallbackExecutor extends SequentialExecutor<CancellableRunnable> {
     CallbackExecutor(Executor executor) {
       super(executor);
     }
@@ -226,19 +221,12 @@ private void cancelQueuedTasks(final String key, Throwable e) {
       // TODO(kimkyung-goog): Ensure execute() fails once cancelQueueTasks() has been ever invoked,
       // so that no more tasks are scheduled.
       synchronized (tasksByKey) {
-        final Deque<Runnable> tasks = tasksByKey.get(key);
+        final Deque<CancellableRunnable> tasks = tasksByKey.get(key);
         if (tasks == null) {
           return;
         }
         while (!tasks.isEmpty()) {
-          Runnable task = tasks.poll();
-          if (task instanceof CancellableRunnable) {
-            ((CancellableRunnable) task).cancel(e);
-          } else {
-            logger.log(
-                Level.WARNING,
-                "Attempted to cancel Runnable that was not CancellableRunnable; ignored.");
-          }
+          tasks.poll().cancel(e);
         }
       }
     }
