Extracted out generic mergePlugins, and cleaned up call sites

Author: donald-pinckney

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java

@@ -17,6 +17,7 @@
 import io.temporal.common.interceptors.WorkflowClientInterceptor;
 import io.temporal.internal.WorkflowThreadMarker;
 import io.temporal.internal.client.*;
+import io.temporal.internal.common.PluginUtils;
 import io.temporal.internal.client.NexusStartWorkflowResponse;
 import io.temporal.internal.client.external.GenericWorkflowClient;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
@@ -75,7 +76,14 @@ public static WorkflowClient newInstance(
         extractClientPlugins(workflowServiceStubs.getOptions().getPlugins());
 
     // Merge propagated plugins with client-specified plugins
-    WorkflowClientPlugin[] mergedPlugins = mergePlugins(propagatedPlugins, options.getPlugins());
+    WorkflowClientPlugin[] mergedPlugins =
+        PluginUtils.mergePlugins(
+            propagatedPlugins,
+            options.getPlugins(),
+            WorkflowClientPlugin::getName,
+            log,
+            "service stubs",
+            WorkflowClientPlugin.class);
 
     // Apply plugin configuration phase (forward order) on user-provided options,
     // so plugins see unmodified state before defaults and plugin merging
@@ -810,37 +818,4 @@ private static WorkflowClientPlugin[] extractClientPlugins(
     }
     return clientPlugins.toArray(new WorkflowClientPlugin[0]);
   }
-
-  /**
-   * Merges propagated plugins with explicitly specified plugins. Propagated plugins come first
-   * (from service stubs), followed by client-specific plugins.
-   */
-  private static WorkflowClientPlugin[] mergePlugins(
-      WorkflowClientPlugin[] propagated, WorkflowClientPlugin[] explicit) {
-    boolean propagatedEmpty = propagated == null || propagated.length == 0;
-    boolean explicitEmpty = explicit == null || explicit.length == 0;
-    if (propagatedEmpty && explicitEmpty) {
-      return new WorkflowClientPlugin[0];
-    }
-    if (propagatedEmpty) {
-      return explicit;
-    }
-    if (explicitEmpty) {
-      return propagated;
-    }
-    // Warn about duplicate plugin instances (same object in both lists)
-    Set<WorkflowClientPlugin> propagatedSet = new HashSet<>(Arrays.asList(propagated));
-    for (WorkflowClientPlugin p : explicit) {
-      if (propagatedSet.contains(p)) {
-        log.warn(
-            "Plugin instance {} is present in both propagated plugins (from service stubs) and "
-                + "explicit plugins. It will run twice which may not be the intended behavior.",
-            p.getName());
-      }
-    }
-    WorkflowClientPlugin[] merged = new WorkflowClientPlugin[propagated.length + explicit.length];
-    System.arraycopy(propagated, 0, merged, 0, propagated.length);
-    System.arraycopy(explicit, 0, merged, propagated.length, explicit.length);
-    return merged;
-  }
 }

temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java

@@ -8,16 +8,14 @@
 import io.temporal.internal.WorkflowThreadMarker;
 import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
 import io.temporal.internal.client.RootScheduleClientInvoker;
+import io.temporal.internal.common.PluginUtils;
 import io.temporal.internal.client.external.GenericWorkflowClient;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.serviceclient.WorkflowServiceStubsPlugin;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
 import org.slf4j.Logger;
@@ -55,7 +53,14 @@ public static ScheduleClient newInstance(
         extractScheduleClientPlugins(workflowServiceStubs.getOptions().getPlugins());
 
     // Merge propagated plugins with schedule client-specified plugins
-    ScheduleClientPlugin[] mergedPlugins = mergePlugins(propagatedPlugins, options.getPlugins());
+    ScheduleClientPlugin[] mergedPlugins =
+        PluginUtils.mergePlugins(
+            propagatedPlugins,
+            options.getPlugins(),
+            ScheduleClientPlugin::getName,
+            log,
+            "service stubs",
+            ScheduleClientPlugin.class);
 
     // Apply plugin configuration phase (forward order) on user-provided options,
     // so plugins see unmodified state before defaults and plugin merging
@@ -95,34 +100,6 @@ private static ScheduleClientPlugin[] extractScheduleClientPlugins(
     return schedulePlugins.toArray(new ScheduleClientPlugin[0]);
   }
 
-  private static ScheduleClientPlugin[] mergePlugins(
-      ScheduleClientPlugin[] propagated, ScheduleClientPlugin[] explicit) {
-    if ((propagated == null || propagated.length == 0)
-        && (explicit == null || explicit.length == 0)) {
-      return new ScheduleClientPlugin[0];
-    }
-    // Warn about duplicate plugin instances (same object in both lists)
-    if (propagated != null && propagated.length > 0 && explicit != null && explicit.length > 0) {
-      Set<ScheduleClientPlugin> propagatedSet = new HashSet<>(Arrays.asList(propagated));
-      for (ScheduleClientPlugin p : explicit) {
-        if (propagatedSet.contains(p)) {
-          log.warn(
-              "Plugin instance {} is present in both propagated plugins (from service stubs) and "
-                  + "explicit plugins. It will run twice which may not be the intended behavior.",
-              p.getName());
-        }
-      }
-    }
-    List<ScheduleClientPlugin> merged = new ArrayList<>();
-    if (propagated != null) {
-      merged.addAll(Arrays.asList(propagated));
-    }
-    if (explicit != null) {
-      merged.addAll(Arrays.asList(explicit));
-    }
-    return merged.toArray(new ScheduleClientPlugin[0]);
-  }
-
   private ScheduleClientCallsInterceptor initializeClientInvoker() {
     ScheduleClientCallsInterceptor scheduleClientInvoker =
         new RootScheduleClientInvoker(genericClient, options);

temporal-sdk/src/main/java/io/temporal/internal/common/PluginUtils.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.internal.common;
+
+import java.lang.reflect.Array;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+
+/** Internal utilities for plugin management. */
+public final class PluginUtils {
+
+  private PluginUtils() {}
+
+  /**
+   * Merges propagated plugins with explicitly specified plugins. Propagated plugins come first,
+   * followed by explicit plugins. Warns about duplicate plugin instances.
+   *
+   * @param propagated plugins propagated from a parent component (may be null or empty)
+   * @param explicit plugins explicitly specified on this component (may be null or empty)
+   * @param nameExtractor function to extract plugin name for logging
+   * @param log logger for duplicate warnings
+   * @param propagatedSource description of where propagated plugins come from (e.g., "service
+   *     stubs", "client")
+   * @param <T> the plugin type
+   * @return merged array of plugins, never null
+   */
+  public static <T> T[] mergePlugins(
+      @Nullable T[] propagated,
+      @Nullable T[] explicit,
+      Function<T, String> nameExtractor,
+      Logger log,
+      String propagatedSource,
+      Class<T> pluginClass) {
+    boolean propagatedEmpty = propagated == null || propagated.length == 0;
+    boolean explicitEmpty = explicit == null || explicit.length == 0;
+
+    if (propagatedEmpty && explicitEmpty) {
+      @SuppressWarnings("unchecked")
+      T[] empty = (T[]) Array.newInstance(pluginClass, 0);
+      return empty;
+    }
+    if (propagatedEmpty) {
+      return explicit;
+    }
+    if (explicitEmpty) {
+      return propagated;
+    }
+
+    // Warn about duplicate plugin instances (same object in both lists)
+    Set<T> propagatedSet = new HashSet<>(Arrays.asList(propagated));
+    for (T p : explicit) {
+      if (propagatedSet.contains(p)) {
+        log.warn(
+            "Plugin instance {} is present in both propagated plugins (from {}) and "
+                + "explicit plugins. It will run twice which may not be the intended behavior.",
+            nameExtractor.apply(p),
+            propagatedSource);
+      }
+    }
+
+    @SuppressWarnings("unchecked")
+    T[] merged = (T[]) Array.newInstance(pluginClass, propagated.length + explicit.length);
+    System.arraycopy(propagated, 0, merged, 0, propagated.length);
+    System.arraycopy(explicit, 0, merged, propagated.length, explicit.length);
+    return merged;
+  }
+}

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

@@ -12,18 +12,17 @@
 import io.temporal.internal.sync.WorkflowThreadExecutor;
 import io.temporal.internal.task.VirtualThreadDelegate;
 import io.temporal.internal.worker.ShutdownManager;
+import io.temporal.internal.common.PluginUtils;
 import io.temporal.internal.worker.WorkflowExecutorCache;
 import io.temporal.internal.worker.WorkflowRunLockManager;
 import io.temporal.serviceclient.MetricsTag;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.SynchronousQueue;
@@ -84,13 +83,21 @@ private WorkerFactory(WorkflowClient workflowClient, WorkerFactoryOptions factor
     String namespace = workflowClientOptions.getNamespace();
 
     // Extract worker plugins from client (auto-propagation)
-    List<WorkerPlugin> propagatedPlugins = extractWorkerPlugins(workflowClientOptions.getPlugins());
+    WorkerPlugin[] propagatedPlugins = extractWorkerPlugins(workflowClientOptions.getPlugins());
 
     // Get plugins explicitly set on factory options
     WorkerPlugin[] explicitPlugins = factoryOptions != null ? factoryOptions.getPlugins() : null;
 
     // Merge propagated plugins with explicit plugins (propagated first)
-    this.plugins = mergePlugins(propagatedPlugins, explicitPlugins);
+    WorkerPlugin[] mergedPlugins =
+        PluginUtils.mergePlugins(
+            propagatedPlugins,
+            explicitPlugins,
+            WorkerPlugin::getName,
+            log,
+            "client",
+            WorkerPlugin.class);
+    this.plugins = Collections.unmodifiableList(Arrays.asList(mergedPlugins));
 
     // Apply plugin configuration to factory options (forward order) on user-provided options,
     // so plugins see unmodified state before defaults and plugin merging
@@ -470,51 +477,18 @@ public String toString() {
    * Extracts worker plugins from the workflow client plugins array. Only plugins that also
    * implement {@link WorkerPlugin} are included.
    */
-  private static List<WorkerPlugin> extractWorkerPlugins(
+  private static WorkerPlugin[] extractWorkerPlugins(
       io.temporal.client.WorkflowClientPlugin[] clientPlugins) {
     if (clientPlugins == null || clientPlugins.length == 0) {
-      return Collections.emptyList();
+      return new WorkerPlugin[0];
     }
-
     List<WorkerPlugin> workerPlugins = new ArrayList<>();
     for (io.temporal.client.WorkflowClientPlugin plugin : clientPlugins) {
       if (plugin instanceof WorkerPlugin) {
         workerPlugins.add((WorkerPlugin) plugin);
       }
     }
-    return Collections.unmodifiableList(workerPlugins);
-  }
-
-  /**
-   * Merges propagated plugins with explicitly specified plugins. Propagated plugins come first
-   * (from client), followed by factory-specific plugins.
-   */
-  private static List<WorkerPlugin> mergePlugins(
-      List<WorkerPlugin> propagated, WorkerPlugin[] explicit) {
-    if ((propagated == null || propagated.isEmpty())
-        && (explicit == null || explicit.length == 0)) {
-      return Collections.emptyList();
-    }
-    if (propagated == null || propagated.isEmpty()) {
-      return Collections.unmodifiableList(Arrays.asList(explicit));
-    }
-    if (explicit == null || explicit.length == 0) {
-      return propagated;
-    }
-    // Warn about duplicate plugin instances (same object in both lists)
-    Set<WorkerPlugin> propagatedSet = new HashSet<>(propagated);
-    for (WorkerPlugin p : explicit) {
-      if (propagatedSet.contains(p)) {
-        log.warn(
-            "Plugin instance {} is present in both propagated plugins (from client) and "
-                + "explicit plugins. It will run twice which may not be the intended behavior.",
-            p.getName());
-      }
-    }
-    List<WorkerPlugin> merged = new ArrayList<>(propagated.size() + explicit.length);
-    merged.addAll(propagated);
-    merged.addAll(Arrays.asList(explicit));
-    return Collections.unmodifiableList(merged);
+    return workerPlugins.toArray(new WorkerPlugin[0]);
   }
 
   /**