first stab at replayer

Author: donald-pinckney

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

@@ -123,6 +123,9 @@ public abstract class SimplePlugin
   private final List<BiConsumer<String, Worker>> workerShutdownCallbacks;
   private final List<Consumer<WorkerFactory>> workerFactoryStartCallbacks;
   private final List<Consumer<WorkerFactory>> workerFactoryShutdownCallbacks;
+  private final List<Consumer<WorkerOptions.Builder>> replayWorkerCustomizers;
+  private final List<BiConsumer<String, Worker>> replayWorkerInitializers;
+  private final List<BiConsumer<Worker, WorkflowExecutionHistory>> replayExecutionCallbacks;
   private final List<WorkerInterceptor> workerInterceptors;
   private final List<WorkflowClientInterceptor> clientInterceptors;
   private final List<ContextPropagator> contextPropagators;
@@ -151,6 +154,9 @@ protected SimplePlugin(@Nonnull String name) {
     this.workerShutdownCallbacks = Collections.emptyList();
     this.workerFactoryStartCallbacks = Collections.emptyList();
     this.workerFactoryShutdownCallbacks = Collections.emptyList();
+    this.replayWorkerCustomizers = Collections.emptyList();
+    this.replayWorkerInitializers = Collections.emptyList();
+    this.replayExecutionCallbacks = Collections.emptyList();
     this.workerInterceptors = Collections.emptyList();
     this.clientInterceptors = Collections.emptyList();
     this.contextPropagators = Collections.emptyList();
@@ -180,6 +186,9 @@ protected SimplePlugin(@Nonnull Builder builder) {
     this.workerShutdownCallbacks = new ArrayList<>(builder.workerShutdownCallbacks);
     this.workerFactoryStartCallbacks = new ArrayList<>(builder.workerFactoryStartCallbacks);
     this.workerFactoryShutdownCallbacks = new ArrayList<>(builder.workerFactoryShutdownCallbacks);
+    this.replayWorkerCustomizers = new ArrayList<>(builder.replayWorkerCustomizers);
+    this.replayWorkerInitializers = new ArrayList<>(builder.replayWorkerInitializers);
+    this.replayExecutionCallbacks = new ArrayList<>(builder.replayExecutionCallbacks);
     this.workerInterceptors = new ArrayList<>(builder.workerInterceptors);
     this.clientInterceptors = new ArrayList<>(builder.clientInterceptors);
     this.contextPropagators = new ArrayList<>(builder.contextPropagators);
@@ -340,6 +349,41 @@ public void shutdownWorkerFactory(WorkerFactory factory, Runnable next) throws E
     next.run();
   }
 
+  @Override
+  public void configureReplayWorker(
+      @Nonnull String taskQueue, @Nonnull WorkerOptions.Builder builder) {
+    if (replayWorkerCustomizers.isEmpty()) {
+      // Default: delegate to configureWorker
+      WorkerPlugin.super.configureReplayWorker(taskQueue, builder);
+    } else {
+      for (Consumer<WorkerOptions.Builder> customizer : replayWorkerCustomizers) {
+        customizer.accept(builder);
+      }
+    }
+  }
+
+  @Override
+  public void initializeReplayWorker(@Nonnull String taskQueue, @Nonnull Worker worker) {
+    if (replayWorkerInitializers.isEmpty()) {
+      // Default: delegate to initializeWorker
+      WorkerPlugin.super.initializeReplayWorker(taskQueue, worker);
+    } else {
+      for (BiConsumer<String, Worker> initializer : replayWorkerInitializers) {
+        initializer.accept(taskQueue, worker);
+      }
+    }
+  }
+
+  @Override
+  public void replayWorkflowExecution(
+      @Nonnull Worker worker, @Nonnull WorkflowExecutionHistory history, @Nonnull Runnable next)
+      throws Exception {
+    next.run();
+    for (BiConsumer<Worker, WorkflowExecutionHistory> callback : replayExecutionCallbacks) {
+      callback.accept(worker, history);
+    }
+  }
+
   @Override
   public String toString() {
     return getClass().getSimpleName() + "{name='" + name + "'}";
@@ -363,6 +407,10 @@ public static final class Builder {
     private final List<BiConsumer<String, Worker>> workerShutdownCallbacks = new ArrayList<>();
     private final List<Consumer<WorkerFactory>> workerFactoryStartCallbacks = new ArrayList<>();
     private final List<Consumer<WorkerFactory>> workerFactoryShutdownCallbacks = new ArrayList<>();
+    private final List<Consumer<WorkerOptions.Builder>> replayWorkerCustomizers = new ArrayList<>();
+    private final List<BiConsumer<String, Worker>> replayWorkerInitializers = new ArrayList<>();
+    private final List<BiConsumer<Worker, WorkflowExecutionHistory>> replayExecutionCallbacks =
+        new ArrayList<>();
     private final List<WorkerInterceptor> workerInterceptors = new ArrayList<>();
     private final List<WorkflowClientInterceptor> clientInterceptors = new ArrayList<>();
     private final List<ContextPropagator> contextPropagators = new ArrayList<>();
@@ -561,6 +609,59 @@ public Builder onWorkerFactoryShutdown(@Nonnull Consumer<WorkerFactory> callback
       return this;
     }
 
+    // ==================== Replay Methods ====================
+
+    /**
+     * Adds a customizer for {@link WorkerOptions} that is applied only when creating replay
+     * workers. If no replay-specific customizers are set, the regular worker customizers are used.
+     *
+     * <p>Use this when replay workers need different configuration than normal workers.
+     *
+     * @param customizer a consumer that modifies the options builder
+     * @return this builder for chaining
+     */
+    public Builder customizeReplayWorker(@Nonnull Consumer<WorkerOptions.Builder> customizer) {
+      replayWorkerCustomizers.add(Objects.requireNonNull(customizer));
+      return this;
+    }
+
+    /**
+     * Adds an initializer that is called after a replay worker is created. If no replay-specific
+     * initializers are set, the regular worker initializers are used.
+     *
+     * <p>Use this when replay workers need different initialization than normal workers.
+     *
+     * @param initializer a consumer that receives the task queue name and worker
+     * @return this builder for chaining
+     */
+    public Builder initializeReplayWorker(@Nonnull BiConsumer<String, Worker> initializer) {
+      replayWorkerInitializers.add(Objects.requireNonNull(initializer));
+      return this;
+    }
+
+    /**
+     * Adds a callback that is invoked after a workflow execution is replayed. This can be used for
+     * logging, metrics, or other observability around replay operations.
+     *
+     * <p>Example:
+     *
+     * <pre>{@code
+     * SimplePlugin.newBuilder("my-plugin")
+     *     .onReplayWorkflowExecution((worker, history) -> {
+     *         logger.info("Replayed workflow: {}", history.getWorkflowExecution().getWorkflowId());
+     *     })
+     *     .build();
+     * }</pre>
+     *
+     * @param callback a consumer that receives the worker and history after replay completes
+     * @return this builder for chaining
+     */
+    public Builder onReplayWorkflowExecution(
+        @Nonnull BiConsumer<Worker, WorkflowExecutionHistory> callback) {
+      replayExecutionCallbacks.add(Objects.requireNonNull(callback));
+      return this;
+    }
+
     /**
      * Adds worker interceptors. Interceptors are appended to any existing interceptors in the
      * configuration.

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

@@ -42,6 +42,7 @@ public final class Worker {
   private static final Logger log = LoggerFactory.getLogger(Worker.class);
   private final WorkerOptions options;
   private final String taskQueue;
+  private final List<WorkerPlugin> plugins;
   final SyncWorkflowWorker workflowWorker;
   final SyncActivityWorker activityWorker;
   final SyncNexusWorker nexusWorker;
@@ -67,9 +68,11 @@ public final class Worker {
       @Nonnull WorkflowExecutorCache cache,
       boolean useStickyTaskQueue,
       WorkflowThreadExecutor workflowThreadExecutor,
-      List<ContextPropagator> contextPropagators) {
+      List<ContextPropagator> contextPropagators,
+      @Nonnull List<WorkerPlugin> plugins) {
 
     Objects.requireNonNull(client, "client should not be null");
+    this.plugins = Objects.requireNonNull(plugins, "plugins should not be null");
     Preconditions.checkArgument(
         !Strings.isNullOrEmpty(taskQueue), "taskQueue should not be an empty string");
     this.taskQueue = taskQueue;
@@ -469,12 +472,57 @@ public String toString() {
   @SuppressWarnings("deprecation")
   public void replayWorkflowExecution(io.temporal.internal.common.WorkflowExecutionHistory history)
       throws Exception {
-    workflowWorker.queryWorkflowExecution(
-        history,
-        WorkflowClient.QUERY_TYPE_REPLAY_ONLY,
-        String.class,
-        String.class,
-        new Object[] {});
+    // Convert to public history type for plugin API
+    WorkflowExecutionHistory publicHistory =
+        new WorkflowExecutionHistory(
+            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 =
+        () -> {
+          try {
+            workflowWorker.queryWorkflowExecution(
+                history,
+                WorkflowClient.QUERY_TYPE_REPLAY_ONLY,
+                String.class,
+                String.class,
+                new Object[] {});
+          } catch (Exception e) {
+            throw new ReplayException(e);
+          }
+        };
+
+    for (int i = plugins.size() - 1; i >= 0; i--) {
+      WorkerPlugin plugin = plugins.get(i);
+      Runnable next = chain;
+      chain =
+          () -> {
+            try {
+              plugin.replayWorkflowExecution(this, publicHistory, next);
+            } catch (Exception e) {
+              throw new ReplayException(e);
+            }
+          };
+    }
+
+    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();
+    }
   }
 
   /**

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

@@ -7,6 +7,7 @@
 import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowClientOptions;
+import io.temporal.common.Experimental;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.internal.client.WorkflowClientInternal;
 import io.temporal.internal.sync.WorkflowThreadExecutor;
@@ -175,7 +176,8 @@ public synchronized Worker newWorker(String taskQueue, WorkerOptions options) {
               cache,
               true,
               workflowThreadExecutor,
-              workflowClient.getOptions().getContextPropagators());
+              workflowClient.getOptions().getContextPropagators(),
+              plugins);
       workers.put(taskQueue, worker);
 
       // Go through the plugins to call plugin initializeWorker hooks (e.g. register workflows,
@@ -194,6 +196,71 @@ public synchronized Worker newWorker(String taskQueue, WorkerOptions options) {
     }
   }
 
+  /**
+   * Creates a worker specifically for replay operations. This method should be used when replaying
+   * workflow histories to ensure plugins receive the replay-specific configuration callbacks.
+   *
+   * <p>Unlike {@link #newWorker(String, WorkerOptions)}, this method:
+   *
+   * <ul>
+   *   <li>Calls {@link WorkerPlugin#configureReplayWorker} instead of {@link
+   *       WorkerPlugin#configureWorker}
+   *   <li>Calls {@link WorkerPlugin#initializeReplayWorker} instead of {@link
+   *       WorkerPlugin#initializeWorker}
+   * </ul>
+   *
+   * <p>This allows plugins to apply different configuration for replay scenarios (e.g., disabling
+   * certain interceptors that shouldn't run during replay).
+   *
+   * @param taskQueue task queue name for the replay worker
+   * @param options Options for configuring the replay worker
+   * @return Worker configured for replay
+   */
+  @Experimental
+  public synchronized Worker newReplayWorker(String taskQueue, WorkerOptions options) {
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(taskQueue), "taskQueue should not be an empty string");
+    Preconditions.checkState(
+        state == State.Initial,
+        String.format(statusErrorMessage, "create new worker", state.name(), State.Initial.name()));
+
+    // Apply replay-specific plugin configuration to worker options (forward order)
+    options = applyReplayWorkerPluginConfiguration(taskQueue, options, this.plugins);
+
+    // Only one worker can exist for a task queue
+    Worker existingWorker = workers.get(taskQueue);
+    if (existingWorker == null) {
+      Worker worker =
+          new Worker(
+              workflowClient,
+              taskQueue,
+              factoryOptions,
+              options,
+              metricsScope,
+              runLocks,
+              cache,
+              true,
+              workflowThreadExecutor,
+              workflowClient.getOptions().getContextPropagators(),
+              plugins);
+      workers.put(taskQueue, worker);
+
+      // Go through the plugins to call plugin initializeReplayWorker hooks (e.g. register
+      // workflows)
+      for (WorkerPlugin plugin : plugins) {
+        plugin.initializeReplayWorker(taskQueue, worker);
+      }
+
+      return worker;
+    } else {
+      log.warn(
+          "Only one worker can be registered for a task queue, "
+              + "subsequent calls to WorkerFactory#newReplayWorker with the same task queue are ignored and "
+              + "initially created worker is returned");
+      return existingWorker;
+    }
+  }
+
   /**
    * @param taskQueue task queue name to lookup an existing worker for
    * @return a worker created previously through {@link #newWorker(String)} for the given task
@@ -598,6 +665,25 @@ private static WorkerOptions applyWorkerPluginConfiguration(
     return builder.build();
   }
 
+  /**
+   * Applies replay-specific plugin configuration to worker options. Plugins are called in forward
+   * (registration) order.
+   */
+  private static WorkerOptions applyReplayWorkerPluginConfiguration(
+      String taskQueue, WorkerOptions options, List<WorkerPlugin> plugins) {
+    if (plugins == null || plugins.isEmpty()) {
+      return options;
+    }
+
+    WorkerOptions.Builder builder =
+        options == null ? WorkerOptions.newBuilder() : WorkerOptions.newBuilder(options);
+
+    for (WorkerPlugin plugin : plugins) {
+      plugin.configureReplayWorker(taskQueue, builder);
+    }
+    return builder.build();
+  }
+
   enum State {
     Initial,
     Started,