Merge branch 'master' into dplyukhin/improve-contributing-guide

Author: dplyukhin

.github/workflows/nightly-throughput-stress.yml

@@ -1,3 +1,5 @@
+![Temporal Java SDK](https://raw.githubusercontent.com/temporalio/assets/main/files/w/java.png)
+
 # Temporal Java SDK  [![Build status](https://github.com/temporalio/sdk-java/actions/workflows/ci.yml/badge.svg?event=push)](https://github.com/temporalio/sdk-java/actions/workflows/ci.yml) [![Coverage Status](https://coveralls.io/repos/github/temporalio/sdk-java/badge.svg?branch=master)](https://coveralls.io/github/temporalio/sdk-java?branch=master)
 
 [Temporal](https://github.com/temporalio/temporal) is a Workflow-as-Code platform for building and operating
@@ -95,4 +97,4 @@ 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.
+limitations under the License.

README.md

@@ -32,11 +32,19 @@ public class LinkConverter {
 
   public static io.temporal.api.nexus.v1.Link workflowEventToNexusLink(Link.WorkflowEvent we) {
     try {
+
       String url =
           String.format(
               linkPathFormat,
               URLEncoder.encode(we.getNamespace(), StandardCharsets.UTF_8.toString()),
-              URLEncoder.encode(we.getWorkflowId(), StandardCharsets.UTF_8.toString()),
+              // The 'replace' below handles spaces - the encoder will convert them to a plus,
+              // which the UI then handles as a plus, thus breaking the link as the
+              // space is lost.
+              // It's a known quirk with the URLEncoder as it encodes for forms, not general URIs.
+              // Only done for the WorkflowId as the other two are values we control,
+              // and will never have spaces.
+              URLEncoder.encode(we.getWorkflowId(), StandardCharsets.UTF_8.toString())
+                  .replace("+", "%20"),
               URLEncoder.encode(we.getRunId(), StandardCharsets.UTF_8.toString()));
 
       List<Map.Entry<String, String>> queryParams = new ArrayList<>();

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

@@ -10,15 +10,21 @@
 public class InternalNexusOperationContext {
   private final String namespace;
   private final String taskQueue;
+  private final String endpoint;
   private final Scope metricScope;
   private final WorkflowClient client;
   NexusOperationOutboundCallsInterceptor outboundCalls;
   Link startWorkflowResponseLink;
 
   public InternalNexusOperationContext(
-      String namespace, String taskQueue, Scope metricScope, WorkflowClient client) {
+      String namespace,
+      String taskQueue,
+      String endpoint,
+      Scope metricScope,
+      WorkflowClient client) {
     this.namespace = namespace;
     this.taskQueue = taskQueue;
+    this.endpoint = endpoint;
     this.metricScope = metricScope;
     this.client = client;
   }
@@ -39,6 +45,10 @@ public String getNamespace() {
     return namespace;
   }
 
+  public String getEndpoint() {
+    return endpoint;
+  }
+
   public void setOutboundInterceptor(NexusOperationOutboundCallsInterceptor outboundCalls) {
     this.outboundCalls = outboundCalls;
   }

temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java

@@ -5,10 +5,12 @@
 class NexusInfoImpl implements NexusOperationInfo {
   private final String namespace;
   private final String taskQueue;
+  private final String endpoint;
 
-  NexusInfoImpl(String namespace, String taskQueue) {
+  NexusInfoImpl(String namespace, String taskQueue, String endpoint) {
     this.namespace = namespace;
     this.taskQueue = taskQueue;
+    this.endpoint = endpoint;
   }
 
   @Override
@@ -20,4 +22,9 @@ public String getNamespace() {
   public String getTaskQueue() {
     return taskQueue;
   }
+
+  @Override
+  public String getEndpoint() {
+    return endpoint;
+  }
 }

temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusInfoImpl.java

@@ -116,7 +116,8 @@ public Result handle(NexusTask task, Scope metricsScope) throws TimeoutException
       }
 
       CurrentNexusOperationContext.set(
-          new InternalNexusOperationContext(namespace, taskQueue, metricsScope, client));
+          new InternalNexusOperationContext(
+              namespace, taskQueue, request.getEndpoint(), metricsScope, client));
 
       switch (request.getVariantCase()) {
         case START_OPERATION:

temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java

@@ -29,7 +29,9 @@ public OperationHandler<Object, Object> intercept(
             temporalNexusContext.getMetricsScope(),
             temporalNexusContext.getWorkflowClient(),
             new NexusInfoImpl(
-                temporalNexusContext.getNamespace(), temporalNexusContext.getTaskQueue())));
+                temporalNexusContext.getNamespace(),
+                temporalNexusContext.getTaskQueue(),
+                temporalNexusContext.getEndpoint())));
     return new OperationInterceptorConverter(inboundCallsInterceptor);
   }
 

temporal-sdk/src/main/java/io/temporal/internal/nexus/TemporalInterceptorMiddleware.java

@@ -21,7 +21,23 @@ public class ShutdownManager implements Closeable {
           new ExecutorThreadFactory(
               WorkerThreadsNameHelper.SHUTDOWN_MANAGER_THREAD_NAME_PREFIX, null));
 
-  private static final int CHECK_PERIOD_MS = 250;
+  private static final int DEFAULT_CHECK_PERIOD_MS = 250;
+  private final int checkPeriodMs;
+
+  public ShutdownManager() {
+    this(DEFAULT_CHECK_PERIOD_MS);
+  }
+
+  /**
+   * @param checkPeriodMs interval in milliseconds between shutdown-completion polls. Lower values
+   *     speed up teardown at the cost of more frequent polling. Must be positive.
+   */
+  public ShutdownManager(int checkPeriodMs) {
+    if (checkPeriodMs <= 0) {
+      throw new IllegalArgumentException("checkPeriodMs must be positive, was: " + checkPeriodMs);
+    }
+    this.checkPeriodMs = checkPeriodMs;
+  }
 
   /** executorToShutdown.shutdownNow() -&gt; timed wait for a graceful termination */
   public CompletableFuture<Void> shutdownExecutorNow(
@@ -97,7 +113,7 @@ private CompletableFuture<Void> untimedWait(
    */
   private CompletableFuture<Void> limitedWait(
       ExecutorService executorToShutdown, String executorName, Duration timeout) {
-    int attempts = (int) Math.ceil((double) timeout.toMillis() / CHECK_PERIOD_MS);
+    int attempts = (int) Math.ceil((double) timeout.toMillis() / checkPeriodMs);
 
     CompletableFuture<Void> future = new CompletableFuture<>();
     scheduledExecutorService.submit(
@@ -167,7 +183,7 @@ public void run() {
         promise.complete(null);
         return;
       }
-      scheduledExecutorService.schedule(this, CHECK_PERIOD_MS, TimeUnit.MILLISECONDS);
+      scheduledExecutorService.schedule(this, checkPeriodMs, TimeUnit.MILLISECONDS);
     }
 
     abstract boolean isTerminated();
@@ -238,7 +254,7 @@ public void run() {
           onSlowTermination();
         }
       }
-      scheduledExecutorService.schedule(this, CHECK_PERIOD_MS, TimeUnit.MILLISECONDS);
+      scheduledExecutorService.schedule(this, checkPeriodMs, TimeUnit.MILLISECONDS);
     }
 
     abstract boolean isTerminated();

temporal-sdk/src/main/java/io/temporal/internal/worker/ShutdownManager.java

@@ -14,4 +14,10 @@ public interface NexusOperationInfo {
    * @return Nexus Task Queue of the worker that is executing the Nexus Operation
    */
   String getTaskQueue();
+
+  /**
+   * @return Endpoint that the Nexus request was addressed to before being forwarded to this worker.
+   *     Supported from server v1.30.0.
+   */
+  String getEndpoint();
 }

temporal-sdk/src/main/java/io/temporal/nexus/NexusOperationInfo.java

@@ -411,7 +411,8 @@ private void shutdownInternal(boolean interruptUserTasks) {
 
   /** Internal method that actually shuts down workers. Called from the plugin chain. */
   private void doShutdown(boolean interruptUserTasks) {
-    ShutdownManager shutdownManager = new ShutdownManager();
+    ShutdownManager shutdownManager =
+        new ShutdownManager((int) factoryOptions.getShutdownCheckInterval().toMillis());
 
     // Shutdown each worker with plugin hooks
     List<CompletableFuture<Void>> shutdownFutures = new ArrayList<>();