Remove checked exceptions and simplify handling of errors

Author: donald-pinckney

temporal-sdk/src/main/java/io/temporal/common/SimplePlugin.java

@@ -41,6 +41,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.Callable;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
@@ -304,8 +305,8 @@ public void initializeWorker(@Nonnull String taskQueue, @Nonnull Worker worker)
   }
 
   @Override
-  public void startWorker(@Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Runnable next)
-      throws Exception {
+  public void startWorker(
+      @Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Runnable next) {
     next.run();
     for (BiConsumer<String, Worker> callback : workerStartCallbacks) {
       callback.accept(taskQueue, worker);
@@ -328,15 +329,15 @@ public WorkflowServiceStubs connectServiceClient(
   }
 
   @Override
-  public void startWorkerFactory(WorkerFactory factory, Runnable next) throws Exception {
+  public void startWorkerFactory(WorkerFactory factory, Runnable next) {
     next.run();
     for (Consumer<WorkerFactory> callback : workerFactoryStartCallbacks) {
       callback.accept(factory);
     }
   }
 
   @Override
-  public void shutdownWorkerFactory(WorkerFactory factory, Runnable next) throws Exception {
+  public void shutdownWorkerFactory(WorkerFactory factory, Runnable next) {
     for (Consumer<WorkerFactory> callback : workerFactoryShutdownCallbacks) {
       callback.accept(factory);
     }
@@ -345,9 +346,11 @@ public void shutdownWorkerFactory(WorkerFactory factory, Runnable next) throws E
 
   @Override
   public void replayWorkflowExecution(
-      @Nonnull Worker worker, @Nonnull WorkflowExecutionHistory history, @Nonnull Runnable next)
+      @Nonnull Worker worker,
+      @Nonnull WorkflowExecutionHistory history,
+      @Nonnull Callable<Void> next)
       throws Exception {
-    next.run();
+    next.call();
     for (BiConsumer<Worker, WorkflowExecutionHistory> callback : replayExecutionCallbacks) {
       callback.accept(worker, history);
     }

temporal-sdk/src/main/java/io/temporal/worker/Worker.java

@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -478,51 +479,28 @@ public void replayWorkflowExecution(io.temporal.internal.common.WorkflowExecutio
             history.getHistory(), history.getWorkflowExecution().getWorkflowId());
 
     // Build plugin chain in reverse order (first plugin wraps all others)
-    // Wrap checked exception in RuntimeException for Runnable compatibility
-    Runnable chain =
+    Callable<Void> chain =
         () -> {
-          try {
-            workflowWorker.queryWorkflowExecution(
-                history,
-                WorkflowClient.QUERY_TYPE_REPLAY_ONLY,
-                String.class,
-                String.class,
-                new Object[] {});
-          } catch (Exception e) {
-            throw new ReplayException(e);
-          }
+          workflowWorker.queryWorkflowExecution(
+              history,
+              WorkflowClient.QUERY_TYPE_REPLAY_ONLY,
+              String.class,
+              String.class,
+              new Object[] {});
+          return null;
         };
 
     for (int i = plugins.size() - 1; i >= 0; i--) {
       WorkerPlugin plugin = plugins.get(i);
-      Runnable next = chain;
+      Callable<Void> next = chain;
       chain =
           () -> {
-            try {
-              plugin.replayWorkflowExecution(this, publicHistory, next);
-            } catch (Exception e) {
-              throw new ReplayException(e);
-            }
+            plugin.replayWorkflowExecution(this, publicHistory, next);
+            return null;
           };
     }
 
-    try {
-      chain.run();
-    } catch (ReplayException e) {
-      throw e.getCause();
-    }
-  }
-
-  /** Internal exception to wrap checked exceptions during replay. */
-  private static class ReplayException extends RuntimeException {
-    ReplayException(Exception cause) {
-      super(cause);
-    }
-
-    @Override
-    public synchronized Exception getCause() {
-      return (Exception) super.getCause();
-    }
+    chain.call();
   }
 
   /**

temporal-sdk/src/main/java/io/temporal/worker/WorkerFactory.java

@@ -255,17 +255,7 @@ public synchronized void start() {
     for (int i = plugins.size() - 1; i >= 0; i--) {
       final Runnable next = startChain;
       final WorkerPlugin workerPlugin = plugins.get(i);
-      startChain =
-          () -> {
-            try {
-              workerPlugin.startWorkerFactory(this, next);
-            } catch (RuntimeException e) {
-              throw e;
-            } catch (Exception e) {
-              throw new RuntimeException(
-                  "Plugin " + workerPlugin.getName() + " failed during startup", e);
-            }
-          };
+      startChain = () -> workerPlugin.startWorkerFactory(this, next);
     }
 
     // Execute the chain
@@ -284,21 +274,7 @@ private void doStart() {
       for (int i = plugins.size() - 1; i >= 0; i--) {
         final Runnable next = startChain;
         final WorkerPlugin workerPlugin = plugins.get(i);
-        startChain =
-            () -> {
-              try {
-                workerPlugin.startWorker(taskQueue, worker, next);
-              } catch (RuntimeException e) {
-                throw e;
-              } catch (Exception e) {
-                throw new RuntimeException(
-                    "Plugin "
-                        + workerPlugin.getName()
-                        + " failed during worker startup for task queue "
-                        + taskQueue,
-                    e);
-              }
-            };
+        startChain = () -> workerPlugin.startWorker(taskQueue, worker, next);
       }
 
       // Execute the chain for this worker
@@ -386,7 +362,7 @@ private void shutdownInternal(boolean interruptUserTasks) {
           () -> {
             try {
               workerPlugin.shutdownWorkerFactory(this, next);
-            } catch (Exception e) {
+            } catch (RuntimeException e) {
               log.warn("Plugin {} failed during shutdown", workerPlugin.getName(), e);
               // Still try to continue shutdown
               next.run();
@@ -423,7 +399,7 @@ private void doShutdown(boolean interruptUserTasks) {
             () -> {
               try {
                 workerPlugin.shutdownWorker(taskQueue, worker, next);
-              } catch (Exception e) {
+              } catch (RuntimeException e) {
                 log.warn(
                     "Plugin {} failed during worker shutdown for task queue {}",
                     workerPlugin.getName(),

temporal-sdk/src/main/java/io/temporal/worker/WorkerPlugin.java

@@ -23,6 +23,7 @@
 import io.temporal.common.Experimental;
 import io.temporal.common.SimplePlugin;
 import io.temporal.common.WorkflowExecutionHistory;
+import java.util.concurrent.Callable;
 import javax.annotation.Nonnull;
 
 /**
@@ -48,7 +49,7 @@
  *     }
  *
  *     @Override
- *     public void startWorkerFactory(WorkerFactory factory, Runnable next) throws Exception {
+ *     public void startWorkerFactory(WorkerFactory factory, Runnable next) {
  *         registry.recordWorkerStart();
  *         try {
  *             next.run();
@@ -127,7 +128,7 @@ public interface WorkerPlugin {
    *
    * <pre>{@code
    * @Override
-   * public void startWorker(String taskQueue, Worker worker, Runnable next) throws Exception {
+   * public void startWorker(String taskQueue, Worker worker, Runnable next) {
    *     logger.info("Starting worker for task queue: {}", taskQueue);
    *     perWorkerResources.put(taskQueue, new ResourcePool());
    *     next.run();
@@ -137,10 +138,8 @@ public interface WorkerPlugin {
    * @param taskQueue the task queue name for the worker
    * @param worker the worker being started
    * @param next runnable that starts the next in chain (eventually starts the actual worker)
-   * @throws Exception if startup fails
    */
-  void startWorker(@Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Runnable next)
-      throws Exception;
+  void startWorker(@Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Runnable next);
 
   /**
    * Allows the plugin to wrap individual worker shutdown. Called during shutdown phase in reverse
@@ -182,7 +181,7 @@ void startWorker(@Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Run
    *
    * <pre>{@code
    * @Override
-   * public void startWorkerFactory(WorkerFactory factory, Runnable next) throws Exception {
+   * public void startWorkerFactory(WorkerFactory factory, Runnable next) {
    *     logger.info("Starting workers...");
    *     next.run();
    *     logger.info("Workers started");
@@ -191,9 +190,8 @@ void startWorker(@Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Run
    *
    * @param factory the worker factory being started
    * @param next runnable that starts the next in chain (eventually starts actual workers)
-   * @throws Exception if startup fails
    */
-  void startWorkerFactory(@Nonnull WorkerFactory factory, @Nonnull Runnable next) throws Exception;
+  void startWorkerFactory(@Nonnull WorkerFactory factory, @Nonnull Runnable next);
 
   /**
    * Allows the plugin to wrap worker factory shutdown. Called during shutdown phase in reverse
@@ -217,8 +215,7 @@ void startWorker(@Nonnull String taskQueue, @Nonnull Worker worker, @Nonnull Run
    * @param factory the worker factory being shut down
    * @param next runnable that shuts down the next in chain (eventually shuts down actual workers)
    */
-  void shutdownWorkerFactory(@Nonnull WorkerFactory factory, @Nonnull Runnable next)
-      throws Exception;
+  void shutdownWorkerFactory(@Nonnull WorkerFactory factory, @Nonnull Runnable next);
 
   // ==================== Replay Methods ====================
 
@@ -234,11 +231,11 @@ void shutdownWorkerFactory(@Nonnull WorkerFactory factory, @Nonnull Runnable nex
    * <pre>{@code
    * @Override
    * public void replayWorkflowExecution(
-   *     Worker worker, WorkflowExecutionHistory history, Runnable next) throws Exception {
+   *     Worker worker, WorkflowExecutionHistory history, Callable<Void> next) throws Exception {
    *     logger.info("Replaying workflow: {}", history.getWorkflowExecution().getWorkflowId());
    *     long start = System.currentTimeMillis();
    *     try {
-   *         next.run();
+   *         next.call();
    *         logger.info("Replay succeeded in {}ms", System.currentTimeMillis() - start);
    *     } catch (Exception e) {
    *         logger.error("Replay failed after {}ms", System.currentTimeMillis() - start, e);
@@ -249,10 +246,12 @@ void shutdownWorkerFactory(@Nonnull WorkerFactory factory, @Nonnull Runnable nex
    *
    * @param worker the worker performing the replay
    * @param history the workflow execution history being replayed
-   * @param next runnable that performs the next in chain (eventually performs the actual replay)
+   * @param next callable that performs the next in chain (eventually performs the actual replay)
    * @throws Exception if replay fails
    */
   void replayWorkflowExecution(
-      @Nonnull Worker worker, @Nonnull WorkflowExecutionHistory history, @Nonnull Runnable next)
+      @Nonnull Worker worker,
+      @Nonnull WorkflowExecutionHistory history,
+      @Nonnull Callable<Void> next)
       throws Exception;
 }

temporal-sdk/src/test/java/io/temporal/common/SimplePluginBuilderTest.java

@@ -541,7 +541,13 @@ public void testOnReplayWorkflowExecution() throws Exception {
     AtomicBoolean nextCalled = new AtomicBoolean(false);
 
     ((WorkerPlugin) plugin)
-        .replayWorkflowExecution(mockWorker, mockHistory, () -> nextCalled.set(true));
+        .replayWorkflowExecution(
+            mockWorker,
+            mockHistory,
+            () -> {
+              nextCalled.set(true);
+              return null;
+            });
 
     assertTrue("next should be called", nextCalled.get());
     assertTrue("Callback should have been called", callbackCalled.get());
@@ -563,7 +569,7 @@ public void testMultipleOnReplayWorkflowExecutionCallbacks() throws Exception {
     Worker mockWorker = mock(Worker.class);
     WorkflowExecutionHistory mockHistory = mock(WorkflowExecutionHistory.class);
 
-    ((WorkerPlugin) plugin).replayWorkflowExecution(mockWorker, mockHistory, () -> {});
+    ((WorkerPlugin) plugin).replayWorkflowExecution(mockWorker, mockHistory, () -> null);
 
     assertEquals("All callbacks should be called", 3, callCount.get());
   }