From 866aacb82761a63b10989cbd10171dba0929da6e Mon Sep 17 00:00:00 2001 From: joshvanl Date: Wed, 11 Mar 2026 17:50:20 +0000 Subject: [PATCH 1/4] feat: add application-level keepalive to prevent ALB idle connection timeouts AWS ALBs do not forward HTTP/2 PING frames, causing idle gRPC connections to be closed. This adds a background loop that periodically calls the existing Hello RPC as application-level traffic to keep the connection alive through L7 load balancers. Signed-off-by: joshvanl --- .../durabletask/DurableTaskGrpcWorker.java | 108 +++++++++++++----- 1 file changed, 79 insertions(+), 29 deletions(-) diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java index 8fae93cf93..5ba88bd9fb 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java @@ -13,6 +13,7 @@ package io.dapr.durabletask; +import com.google.protobuf.Empty; import io.dapr.durabletask.implementation.protobuf.Orchestration; import io.dapr.durabletask.implementation.protobuf.OrchestratorService; import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; @@ -36,6 +37,7 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; @@ -49,6 +51,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { private static final int DEFAULT_PORT = 4001; private static final Logger logger = Logger.getLogger(DurableTaskGrpcWorker.class.getPackage().getName()); private static final Duration DEFAULT_MAXIMUM_TIMER_INTERVAL = Duration.ofDays(3); + private static final long KEEPALIVE_INTERVAL_SECONDS = 30; private final TaskOrchestrationFactories orchestrationFactories; @@ -64,6 +67,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient; private final boolean isExecutorServiceManaged; private volatile boolean isNormalShutdown = false; + private ScheduledExecutorService keepaliveScheduler; private volatile Thread workerThread; DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder) { @@ -137,6 +141,7 @@ public void close() { this.workerThread.interrupt(); } this.isNormalShutdown = true; + this.stopKeepaliveLoop(); this.shutDownWorkerPool(); this.closeSideCarChannel(); } @@ -177,36 +182,41 @@ public void startAndBlock() { OrchestratorService.GetWorkItemsRequest getWorkItemsRequest = OrchestratorService.GetWorkItemsRequest .newBuilder().build(); Iterator workItemStream = this.sidecarClient.getWorkItems(getWorkItemsRequest); - while (workItemStream.hasNext()) { - if (this.isNormalShutdown || Thread.currentThread().isInterrupted()) { - break; - } - OrchestratorService.WorkItem workItem = workItemStream.next(); - OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase(); - - if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { - OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); - logger.log(Level.FINEST, - String.format("Processing orchestrator request for instance: {0}", - orchestratorRequest.getInstanceId())); - - this.workerPool.submit(new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); - } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { - OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); - - logger.log(Level.INFO, - String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s", - activityRequest.getName(), - activityRequest.getWorkflowInstance().getInstanceId(), - Context.current())); - - this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer)); - - } else { - logger.log(Level.WARNING, - "Received and dropped an unknown '{0}' work-item from the sidecar.", - requestType); + startKeepaliveLoop(); + try { + while (workItemStream.hasNext()) { + if (this.isNormalShutdown || Thread.currentThread().isInterrupted()) { + break; + } + OrchestratorService.WorkItem workItem = workItemStream.next(); + OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase(); + + if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { + OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); + logger.log(Level.FINEST, + String.format("Processing orchestrator request for instance: {0}", + orchestratorRequest.getInstanceId())); + + this.workerPool.submit(new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); + } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { + OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); + + logger.log(Level.INFO, + String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s", + activityRequest.getName(), + activityRequest.getWorkflowInstance().getInstanceId(), + Context.current())); + + this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer)); + + } else { + logger.log(Level.WARNING, + "Received and dropped an unknown '{0}' work-item from the sidecar.", + requestType); + } } + } finally { + stopKeepaliveLoop(); } } catch (StatusRuntimeException e) { if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) { @@ -282,6 +292,46 @@ private void shutDownWorkerPool() { } } + /** + * Starts a background keepalive loop to keep the gRPC connection alive. + * This is an application-level keepalive to prevent AWS ALBs from + * killing idle HTTP/2 connections. + */ + private synchronized void startKeepaliveLoop() { + stopKeepaliveLoop(); + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "durabletask-keepalive"); + t.setDaemon(true); + return t; + }); + try { + scheduler.scheduleWithFixedDelay(() -> { + try { + this.sidecarClient + .withDeadlineAfter(5, TimeUnit.SECONDS) + .hello(Empty.getDefaultInstance()); + } catch (StatusRuntimeException e) { + logger.log(Level.FINE, "keepalive failed", e); + } + }, KEEPALIVE_INTERVAL_SECONDS, KEEPALIVE_INTERVAL_SECONDS, TimeUnit.SECONDS); + } catch (RuntimeException e) { + scheduler.shutdownNow(); + throw e; + } + this.keepaliveScheduler = scheduler; + } + + /** + * Stops the background keepalive loop if one is running. + */ + private synchronized void stopKeepaliveLoop() { + ScheduledExecutorService scheduler = this.keepaliveScheduler; + this.keepaliveScheduler = null; + if (scheduler != null) { + scheduler.shutdownNow(); + } + } + private String getSidecarAddress() { return this.sidecarClient.getChannel().authority(); } From 42d97987d097a169b58e56f045150783482a7e1a Mon Sep 17 00:00:00 2001 From: joshvanl Date: Wed, 11 Mar 2026 18:40:43 +0000 Subject: [PATCH 2/4] Fix Signed-off-by: joshvanl --- .../main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java index 5ba88bd9fb..84e0cf408d 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java @@ -197,7 +197,8 @@ public void startAndBlock() { String.format("Processing orchestrator request for instance: {0}", orchestratorRequest.getInstanceId())); - this.workerPool.submit(new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); + this.workerPool.submit( + new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); From c0f22a25157d00affdbb784c1ac992f0a866ddd5 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Thu, 30 Jul 2026 09:58:48 +0200 Subject: [PATCH 3/4] refactor: move app-level keepalive to workflow channel owners, behind opt-in property Relocate the hello() keepalive out of DurableTaskGrpcWorker (reverted to master) into sdk-workflows, per review feedback: the keepalive is a channel concern, so it now lives with the classes that create workflow channels. - New internal GrpcChannelKeepalive: pings hello() on a daemon thread, catches Throwable so a ping failure can never cancel the periodic task, applies the 5s deadline per ping. - WorkflowRuntimeBuilder creates it on the worker channel; WorkflowRuntime owns and closes it (new optional constructor arg). - DaprWorkflowClient wires it on the client channel, protecting long waitForInstanceCompletion calls through L7 load balancers too. - Gated behind dapr.workflows.app.keep.alive.enabled (default false) with dapr.workflows.app.keep.alive.interval.seconds (default 30), following the existing opt-in convention of dapr.grpc.enable.keep.alive. - Tests: in-process gRPC server coverage for ping cadence, failure tolerance, and close semantics; gating tests for both channel owners. Signed-off-by: Javier Aliaga --- .../durabletask/DurableTaskGrpcWorker.java | 109 ++++---------- sdk-workflows/pom.xml | 5 + .../workflows/client/DaprWorkflowClient.java | 23 ++- .../internal/GrpcChannelKeepalive.java | 78 ++++++++++ .../workflows/runtime/WorkflowRuntime.java | 24 ++++ .../runtime/WorkflowRuntimeBuilder.java | 10 +- .../DaprWorkflowClientKeepaliveTest.java | 59 ++++++++ .../client/DaprWorkflowClientTest.java | 4 +- .../internal/GrpcChannelKeepaliveTest.java | 134 ++++++++++++++++++ .../runtime/WorkflowRuntimeTest.java | 25 ++++ .../main/java/io/dapr/config/Properties.java | 25 ++++ 11 files changed, 408 insertions(+), 88 deletions(-) create mode 100644 sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java create mode 100644 sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java create mode 100644 sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java index 84e0cf408d..8fae93cf93 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java @@ -13,7 +13,6 @@ package io.dapr.durabletask; -import com.google.protobuf.Empty; import io.dapr.durabletask.implementation.protobuf.Orchestration; import io.dapr.durabletask.implementation.protobuf.OrchestratorService; import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; @@ -37,7 +36,6 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; @@ -51,7 +49,6 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { private static final int DEFAULT_PORT = 4001; private static final Logger logger = Logger.getLogger(DurableTaskGrpcWorker.class.getPackage().getName()); private static final Duration DEFAULT_MAXIMUM_TIMER_INTERVAL = Duration.ofDays(3); - private static final long KEEPALIVE_INTERVAL_SECONDS = 30; private final TaskOrchestrationFactories orchestrationFactories; @@ -67,7 +64,6 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient; private final boolean isExecutorServiceManaged; private volatile boolean isNormalShutdown = false; - private ScheduledExecutorService keepaliveScheduler; private volatile Thread workerThread; DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder) { @@ -141,7 +137,6 @@ public void close() { this.workerThread.interrupt(); } this.isNormalShutdown = true; - this.stopKeepaliveLoop(); this.shutDownWorkerPool(); this.closeSideCarChannel(); } @@ -182,42 +177,36 @@ public void startAndBlock() { OrchestratorService.GetWorkItemsRequest getWorkItemsRequest = OrchestratorService.GetWorkItemsRequest .newBuilder().build(); Iterator workItemStream = this.sidecarClient.getWorkItems(getWorkItemsRequest); - startKeepaliveLoop(); - try { - while (workItemStream.hasNext()) { - if (this.isNormalShutdown || Thread.currentThread().isInterrupted()) { - break; - } - OrchestratorService.WorkItem workItem = workItemStream.next(); - OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase(); - - if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { - OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); - logger.log(Level.FINEST, - String.format("Processing orchestrator request for instance: {0}", - orchestratorRequest.getInstanceId())); - - this.workerPool.submit( - new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); - } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { - OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); - - logger.log(Level.INFO, - String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s", - activityRequest.getName(), - activityRequest.getWorkflowInstance().getInstanceId(), - Context.current())); - - this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer)); - - } else { - logger.log(Level.WARNING, - "Received and dropped an unknown '{0}' work-item from the sidecar.", - requestType); - } + while (workItemStream.hasNext()) { + if (this.isNormalShutdown || Thread.currentThread().isInterrupted()) { + break; + } + OrchestratorService.WorkItem workItem = workItemStream.next(); + OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase(); + + if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { + OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); + logger.log(Level.FINEST, + String.format("Processing orchestrator request for instance: {0}", + orchestratorRequest.getInstanceId())); + + this.workerPool.submit(new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); + } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { + OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); + + logger.log(Level.INFO, + String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s", + activityRequest.getName(), + activityRequest.getWorkflowInstance().getInstanceId(), + Context.current())); + + this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer)); + + } else { + logger.log(Level.WARNING, + "Received and dropped an unknown '{0}' work-item from the sidecar.", + requestType); } - } finally { - stopKeepaliveLoop(); } } catch (StatusRuntimeException e) { if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) { @@ -293,46 +282,6 @@ private void shutDownWorkerPool() { } } - /** - * Starts a background keepalive loop to keep the gRPC connection alive. - * This is an application-level keepalive to prevent AWS ALBs from - * killing idle HTTP/2 connections. - */ - private synchronized void startKeepaliveLoop() { - stopKeepaliveLoop(); - ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> { - Thread t = new Thread(r, "durabletask-keepalive"); - t.setDaemon(true); - return t; - }); - try { - scheduler.scheduleWithFixedDelay(() -> { - try { - this.sidecarClient - .withDeadlineAfter(5, TimeUnit.SECONDS) - .hello(Empty.getDefaultInstance()); - } catch (StatusRuntimeException e) { - logger.log(Level.FINE, "keepalive failed", e); - } - }, KEEPALIVE_INTERVAL_SECONDS, KEEPALIVE_INTERVAL_SECONDS, TimeUnit.SECONDS); - } catch (RuntimeException e) { - scheduler.shutdownNow(); - throw e; - } - this.keepaliveScheduler = scheduler; - } - - /** - * Stops the background keepalive loop if one is running. - */ - private synchronized void stopKeepaliveLoop() { - ScheduledExecutorService scheduler = this.keepaliveScheduler; - this.keepaliveScheduler = null; - if (scheduler != null) { - scheduler.shutdownNow(); - } - } - private String getSidecarAddress() { return this.sidecarClient.getChannel().authority(); } diff --git a/sdk-workflows/pom.xml b/sdk-workflows/pom.xml index 0d52f5d9b1..274c989424 100644 --- a/sdk-workflows/pom.xml +++ b/sdk-workflows/pom.xml @@ -46,6 +46,11 @@ io.opentelemetry opentelemetry-api + + io.grpc + grpc-testing + test + diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java index d97b1e288b..f817a8b447 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java @@ -22,6 +22,7 @@ import io.dapr.utils.NetworkUtils; import io.dapr.workflows.Workflow; import io.dapr.workflows.internal.ApiTokenClientInterceptor; +import io.dapr.workflows.internal.GrpcChannelKeepalive; import io.dapr.workflows.runtime.DefaultWorkflowInstanceStatus; import io.dapr.workflows.runtime.DefaultWorkflowState; import io.grpc.ClientInterceptor; @@ -45,6 +46,7 @@ public class DaprWorkflowClient implements AutoCloseable { private ClientInterceptor workflowApiTokenInterceptor; private DurableTaskClient innerClient; private ManagedChannel grpcChannel; + private GrpcChannelKeepalive keepalive; /** * Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel. @@ -75,7 +77,8 @@ public DaprWorkflowClient(Properties properties) { * and this constructor behaves exactly like {@link #DaprWorkflowClient(Properties)}. */ public DaprWorkflowClient(Properties properties, @Nullable Tracer tracer) { - this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)), tracer); + this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)), + tracer, properties); } /** @@ -87,7 +90,7 @@ public DaprWorkflowClient(Properties properties, @Nullable Tracer tracer) { * @param additionalInterceptors extra interceptors appended after the API-token interceptor. */ protected DaprWorkflowClient(Properties properties, ClientInterceptor... additionalInterceptors) { - this(buildChannelWithAdditional(properties, additionalInterceptors), null); + this(buildChannelWithAdditional(properties, additionalInterceptors), null, properties); } /** @@ -95,9 +98,10 @@ protected DaprWorkflowClient(Properties properties, ClientInterceptor... additio * * @param grpcChannel ManagedChannel for GRPC channel. * @param tracer optional Tracer used to propagate trace context when scheduling workflows. + * @param properties Properties the channel was built from. */ - private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) { - this(createDurableTaskClient(grpcChannel, tracer), grpcChannel); + private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer, Properties properties) { + this(createDurableTaskClient(grpcChannel, tracer), grpcChannel, properties); } /** @@ -105,10 +109,15 @@ private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) * * @param innerClient DurableTaskGrpcClient with GRPC Channel set up. * @param grpcChannel ManagedChannel for instance variable setting. + * @param properties Properties the channel was built from. */ - private DaprWorkflowClient(DurableTaskClient innerClient, ManagedChannel grpcChannel) { + private DaprWorkflowClient(DurableTaskClient innerClient, ManagedChannel grpcChannel, Properties properties) { this.innerClient = innerClient; this.grpcChannel = grpcChannel; + if (properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_ENABLED)) { + this.keepalive = new GrpcChannelKeepalive(grpcChannel, "dapr-workflow-client-keepalive", + properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS)); + } } /** @@ -441,6 +450,10 @@ public boolean purgeWorkflow(String workflowInstanceId) { * Closes the inner DurableTask client and shutdown the GRPC channel. */ public void close() throws InterruptedException { + if (this.keepalive != null) { + this.keepalive.close(); + this.keepalive = null; + } try { if (this.innerClient != null) { this.innerClient.close(); diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java b/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java new file mode 100644 index 0000000000..e9738e9b01 --- /dev/null +++ b/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java @@ -0,0 +1,78 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file 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.dapr.workflows.internal; + +import com.google.protobuf.Empty; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * Application-level keepalive for a gRPC channel to the Dapr sidecar. + * + *

Periodically invokes the sidecar's {@code hello} RPC so that intermediaries + * (e.g. AWS ALBs) that do not treat HTTP/2 PING frames as connection activity + * never see the connection as idle and close it.

+ */ +public final class GrpcChannelKeepalive implements AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(GrpcChannelKeepalive.class); + private static final long KEEPALIVE_DEADLINE_SECONDS = 5; + + private final ScheduledExecutorService scheduler; + + /** + * Starts a keepalive loop on the given channel. + * + * @param channel channel to the Dapr sidecar to keep alive. + * @param threadName name of the keepalive thread, to tell instances apart in thread dumps. + * @param interval delay between pings. + */ + public GrpcChannelKeepalive(Channel channel, String threadName, Duration interval) { + TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub stub = + TaskHubSidecarServiceGrpc.newBlockingStub(channel); + this.scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, threadName); + thread.setDaemon(true); + return thread; + }); + // The deadline must be applied per ping: Deadline.after snapshots the clock when + // withDeadlineAfter is invoked, so hoisting it onto the cached stub would make + // every ping after the first fail immediately with DEADLINE_EXCEEDED. + // Catch Throwable, not just RuntimeException: any throwable escaping a periodic + // task silently cancels all future runs. + this.scheduler.scheduleWithFixedDelay(() -> { + try { + stub.withDeadlineAfter(KEEPALIVE_DEADLINE_SECONDS, TimeUnit.SECONDS) + .hello(Empty.getDefaultInstance()); + } catch (Throwable e) { + LOGGER.debug("Sidecar keepalive ping failed", e); + } + }, interval.toMillis(), interval.toMillis(), TimeUnit.MILLISECONDS); + } + + /** + * Stops the keepalive loop. + */ + @Override + public void close() { + this.scheduler.shutdownNow(); + } +} diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java index f5d47f3e79..1e6742210a 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java @@ -14,8 +14,11 @@ package io.dapr.workflows.runtime; import io.dapr.durabletask.DurableTaskGrpcWorker; +import io.dapr.workflows.internal.GrpcChannelKeepalive; import io.grpc.ManagedChannel; +import javax.annotation.Nullable; + import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; @@ -27,6 +30,7 @@ public class WorkflowRuntime implements AutoCloseable { private final DurableTaskGrpcWorker worker; private final ManagedChannel managedChannel; private final ExecutorService executorService; + private final GrpcChannelKeepalive keepalive; /** * Constructor. @@ -38,9 +42,26 @@ public class WorkflowRuntime implements AutoCloseable { public WorkflowRuntime(DurableTaskGrpcWorker worker, ManagedChannel managedChannel, ExecutorService executorService) { + this(worker, managedChannel, executorService, null); + } + + /** + * Constructor. + * + * @param worker grpcWorker processing activities. + * @param managedChannel grpc channel. + * @param executorService executor service responsible for running the threads. + * @param keepalive application-level keepalive on the worker's channel, stopped on + * {@link #close()}. May be null when no keepalive is wanted. + */ + public WorkflowRuntime(DurableTaskGrpcWorker worker, + ManagedChannel managedChannel, + ExecutorService executorService, + @Nullable GrpcChannelKeepalive keepalive) { this.worker = worker; this.managedChannel = managedChannel; this.executorService = executorService; + this.keepalive = keepalive; } /** @@ -68,6 +89,9 @@ public void start(boolean block) { * {@inheritDoc} */ public void close() { + if (this.keepalive != null) { + this.keepalive.close(); + } this.shutDownWorkerPool(); this.closeSideCarChannel(); this.worker.close(); diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java index 3f901e7b0e..371eb01bdb 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java @@ -21,6 +21,7 @@ import io.dapr.workflows.Workflow; import io.dapr.workflows.WorkflowActivity; import io.dapr.workflows.internal.ApiTokenClientInterceptor; +import io.dapr.workflows.internal.GrpcChannelKeepalive; import io.grpc.ClientInterceptor; import io.grpc.ManagedChannel; import org.apache.commons.lang3.StringUtils; @@ -43,6 +44,7 @@ public class WorkflowRuntimeBuilder { private final Set workflowSet = Collections.synchronizedSet(new HashSet<>()); private final DurableTaskGrpcWorkerBuilder builder; private final ManagedChannel managedChannel; + private final Properties properties; private ExecutorService executorService; /** @@ -69,6 +71,7 @@ private WorkflowRuntimeBuilder(Properties properties, Logger logger) { this.workflowApiTokenInterceptor = new ApiTokenClientInterceptor(properties); this.managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, workflowApiTokenInterceptor); this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(this.managedChannel); + this.properties = properties; this.logger = logger; } @@ -82,9 +85,14 @@ public WorkflowRuntime build() { synchronized (WorkflowRuntime.class) { this.executorService = this.executorService == null ? Executors.newCachedThreadPool() : this.executorService; if (instance == null) { + GrpcChannelKeepalive keepalive = null; + if (this.properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_ENABLED)) { + keepalive = new GrpcChannelKeepalive(this.managedChannel, "dapr-workflow-runtime-keepalive", + this.properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS)); + } instance = new WorkflowRuntime( this.builder.withExecutorService(this.executorService).build(), - this.managedChannel, this.executorService); + this.managedChannel, this.executorService, keepalive); } } } diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java new file mode 100644 index 0000000000..a0283a873f --- /dev/null +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java @@ -0,0 +1,59 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file 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.dapr.workflows.client; + +import io.dapr.config.Properties; +import org.junit.jupiter.api.Test; + +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class DaprWorkflowClientKeepaliveTest { + + private static final String THREAD_NAME = "dapr-workflow-client-keepalive"; + + @Test + public void keepaliveIsOffByDefault() throws InterruptedException { + DaprWorkflowClient client = new DaprWorkflowClient(new Properties()); + try { + assertFalse(keepaliveThreadAlive(), "no keepalive thread expected when the property is off"); + } finally { + client.close(); + } + } + + @Test + public void keepaliveStartsWhenEnabledAndStopsOnClose() throws InterruptedException { + Properties properties = new Properties(Map.of("dapr.workflows.app.keep.alive.enabled", "true")); + DaprWorkflowClient client = new DaprWorkflowClient(properties); + try { + assertTrue(keepaliveThreadAlive(), "keepalive thread expected when the property is on"); + } finally { + client.close(); + } + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (keepaliveThreadAlive() && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertFalse(keepaliveThreadAlive(), "keepalive thread should terminate after close()"); + } + + private static boolean keepaliveThreadAlive() { + return Thread.getAllStackTraces().keySet().stream() + .anyMatch(t -> t.getName().equals(THREAD_NAME) && t.isAlive()); + } +} diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java index f88a7dbcc0..fa07479f8d 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java @@ -70,7 +70,7 @@ public WorkflowStub create() { public static void beforeAll() { constructor = Constructor.class.cast(Arrays.stream(DaprWorkflowClient.class.getDeclaredConstructors()) - .filter(c -> c.getParameters().length == 2 + .filter(c -> c.getParameters().length == 3 && c.getParameterTypes()[0] == DurableTaskClient.class) .map(c -> { c.setAccessible(true); @@ -85,7 +85,7 @@ public void setUp() throws Exception { when(mockGrpcChannel.shutdown()).thenReturn(mockGrpcChannel); - client = constructor.newInstance(mockInnerClient, mockGrpcChannel); + client = constructor.newInstance(mockInnerClient, mockGrpcChannel, new Properties()); } @Test diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java new file mode 100644 index 0000000000..98ec321439 --- /dev/null +++ b/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java @@ -0,0 +1,134 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file 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.dapr.workflows.internal; + +import com.google.protobuf.Empty; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class GrpcChannelKeepaliveTest { + + private static final Duration TEST_INTERVAL = Duration.ofMillis(50); + + private final AtomicInteger helloCount = new AtomicInteger(); + private final AtomicBoolean failPings = new AtomicBoolean(false); + + private Server server; + private ManagedChannel channel; + + @BeforeEach + public void setUp() throws IOException { + String serverName = InProcessServerBuilder.generateName(); + this.server = InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void hello(Empty request, StreamObserver responseObserver) { + helloCount.incrementAndGet(); + if (failPings.get()) { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + return; + } + responseObserver.onNext(Empty.getDefaultInstance()); + responseObserver.onCompleted(); + } + }) + .build() + .start(); + this.channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + } + + @AfterEach + public void tearDown() { + this.channel.shutdownNow(); + this.server.shutdownNow(); + } + + @Test + public void pingsPeriodically() throws InterruptedException { + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + awaitHelloCountAtLeast(2); + } + } + + @Test + public void continuesPingingAfterFailures() throws InterruptedException { + failPings.set(true); + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + awaitHelloCountAtLeast(2); + failPings.set(false); + awaitHelloCountAtLeast(helloCount.get() + 2); + } + } + + @Test + public void closeStopsPinging() throws InterruptedException { + GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL); + awaitHelloCountAtLeast(1); + keepalive.close(); + // Allow an already in-flight ping to finish before snapshotting. + Thread.sleep(TEST_INTERVAL.toMillis() * 2); + int countAfterClose = helloCount.get(); + Thread.sleep(TEST_INTERVAL.toMillis() * 4); + assertEquals(countAfterClose, helloCount.get()); + } + + @Test + public void keepaliveThreadIsDaemonAndStopsOnClose() throws InterruptedException { + String threadName = "test-keepalive-lifecycle"; + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, TEST_INTERVAL)) { + awaitHelloCountAtLeast(1); + Thread thread = findThread(threadName); + assertTrue(thread != null && thread.isDaemon(), "expected a live daemon keepalive thread"); + } + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (findThread(threadName) != null && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertTrue(findThread(threadName) == null, "keepalive thread should terminate after close()"); + } + + private void awaitHelloCountAtLeast(int expected) throws InterruptedException { + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (helloCount.get() < expected && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertTrue(helloCount.get() >= expected, + "expected at least " + expected + " hello pings, got " + helloCount.get()); + } + + private static Thread findThread(String name) { + return Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().equals(name) && t.isAlive()) + .findFirst() + .orElse(null); + } +} diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java index 9ee2710e79..2eefcfe527 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java @@ -18,12 +18,17 @@ import io.dapr.durabletask.DurableTaskGrpcWorkerBuilder; import io.dapr.config.Properties; import io.dapr.utils.NetworkUtils; +import io.dapr.workflows.internal.GrpcChannelKeepalive; +import io.grpc.ManagedChannel; import org.junit.jupiter.api.Test; +import java.time.Duration; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertFalse; public class WorkflowRuntimeTest { @@ -35,6 +40,26 @@ public void startTest() { assertDoesNotThrow(() -> runtime.start(false)); } + @Test + public void closeStopsKeepalive() throws InterruptedException { + String threadName = "dapr-workflow-runtime-keepalive"; + DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build(); + ManagedChannel channel = NetworkUtils.buildGrpcManagedChannel(new Properties()); + GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, Duration.ofSeconds(30)); + WorkflowRuntime runtime = new WorkflowRuntime(worker, channel, Executors.newCachedThreadPool(), keepalive); + assertDoesNotThrow(runtime::close); + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (keepaliveThreadAlive(threadName) && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertFalse(keepaliveThreadAlive(threadName), "keepalive thread should terminate after close()"); + } + + private static boolean keepaliveThreadAlive(String threadName) { + return Thread.getAllStackTraces().keySet().stream() + .anyMatch(t -> t.getName().equals(threadName) && t.isAlive()); + } + @Test public void closeWithoutStarting() { DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build(); diff --git a/sdk/src/main/java/io/dapr/config/Properties.java b/sdk/src/main/java/io/dapr/config/Properties.java index 85110c1922..9f5060bac8 100644 --- a/sdk/src/main/java/io/dapr/config/Properties.java +++ b/sdk/src/main/java/io/dapr/config/Properties.java @@ -189,6 +189,31 @@ public class Properties { "DAPR_GRPC_KEEP_ALIVE_WITHOUT_CALLS", true); + /** + * Enables the application-level keepalive on workflow gRPC channels. + * When enabled, the SDK periodically invokes the sidecar's hello RPC so that + * intermediaries that do not treat HTTP/2 PING frames as connection activity + * (e.g. AWS ALBs) never see the connection as idle and close it. + * Environment variable: DAPR_WORKFLOWS_APP_KEEP_ALIVE_ENABLED + * System property: dapr.workflows.app.keep.alive.enabled + * Default: false + */ + public static final Property WORKFLOWS_APP_KEEP_ALIVE_ENABLED = new BooleanProperty( + "dapr.workflows.app.keep.alive.enabled", + "DAPR_WORKFLOWS_APP_KEEP_ALIVE_ENABLED", + false); + + /** + * Interval between application-level keepalive pings on workflow gRPC channels. + * Environment variable: DAPR_WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS + * System property: dapr.workflows.app.keep.alive.interval.seconds + * Default: 30 seconds + */ + public static final Property WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS = new SecondsDurationProperty( + "dapr.workflows.app.keep.alive.interval.seconds", + "DAPR_WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS", + Duration.ofSeconds(30)); + /** * GRPC endpoint for remote sidecar connectivity. */ From 811eee75ed0896507624b5341fbbc36eca13e2d5 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Thu, 30 Jul 2026 10:58:57 +0200 Subject: [PATCH 4/4] feat: enable runtime keepalive by default, scoped to start/close Restore fix-by-default for the worker channel: the keepalive now defaults to on, but only pings between WorkflowRuntime.start() and close() - a started runtime is by definition talking to a sidecar, so idle apps and built-but-unstarted runtimes send nothing. - GrpcChannelKeepalive lifecycle is explicit: inert constructor, idempotent start() and close(). - dapr.workflows.app.keep.alive.enabled (opt-in) is replaced by dapr.workflows.runtime.app.keep.alive.enabled (default true, opt-out). - Drop the client-channel keepalive: a dead idle client connection reconnects on the next call, so it isn't worth the extra surface; DaprWorkflowClient reverts to master. Signed-off-by: Javier Aliaga --- .../workflows/client/DaprWorkflowClient.java | 23 ++------ .../internal/GrpcChannelKeepalive.java | 41 +++++++++---- .../workflows/runtime/WorkflowRuntime.java | 8 ++- .../runtime/WorkflowRuntimeBuilder.java | 2 +- .../DaprWorkflowClientKeepaliveTest.java | 59 ------------------- .../client/DaprWorkflowClientTest.java | 4 +- .../internal/GrpcChannelKeepaliveTest.java | 14 +++++ .../runtime/WorkflowRuntimeTest.java | 6 +- .../main/java/io/dapr/config/Properties.java | 22 +++---- 9 files changed, 73 insertions(+), 106 deletions(-) delete mode 100644 sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java index f817a8b447..d97b1e288b 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java @@ -22,7 +22,6 @@ import io.dapr.utils.NetworkUtils; import io.dapr.workflows.Workflow; import io.dapr.workflows.internal.ApiTokenClientInterceptor; -import io.dapr.workflows.internal.GrpcChannelKeepalive; import io.dapr.workflows.runtime.DefaultWorkflowInstanceStatus; import io.dapr.workflows.runtime.DefaultWorkflowState; import io.grpc.ClientInterceptor; @@ -46,7 +45,6 @@ public class DaprWorkflowClient implements AutoCloseable { private ClientInterceptor workflowApiTokenInterceptor; private DurableTaskClient innerClient; private ManagedChannel grpcChannel; - private GrpcChannelKeepalive keepalive; /** * Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel. @@ -77,8 +75,7 @@ public DaprWorkflowClient(Properties properties) { * and this constructor behaves exactly like {@link #DaprWorkflowClient(Properties)}. */ public DaprWorkflowClient(Properties properties, @Nullable Tracer tracer) { - this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)), - tracer, properties); + this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)), tracer); } /** @@ -90,7 +87,7 @@ public DaprWorkflowClient(Properties properties, @Nullable Tracer tracer) { * @param additionalInterceptors extra interceptors appended after the API-token interceptor. */ protected DaprWorkflowClient(Properties properties, ClientInterceptor... additionalInterceptors) { - this(buildChannelWithAdditional(properties, additionalInterceptors), null, properties); + this(buildChannelWithAdditional(properties, additionalInterceptors), null); } /** @@ -98,10 +95,9 @@ protected DaprWorkflowClient(Properties properties, ClientInterceptor... additio * * @param grpcChannel ManagedChannel for GRPC channel. * @param tracer optional Tracer used to propagate trace context when scheduling workflows. - * @param properties Properties the channel was built from. */ - private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer, Properties properties) { - this(createDurableTaskClient(grpcChannel, tracer), grpcChannel, properties); + private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) { + this(createDurableTaskClient(grpcChannel, tracer), grpcChannel); } /** @@ -109,15 +105,10 @@ private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer, * * @param innerClient DurableTaskGrpcClient with GRPC Channel set up. * @param grpcChannel ManagedChannel for instance variable setting. - * @param properties Properties the channel was built from. */ - private DaprWorkflowClient(DurableTaskClient innerClient, ManagedChannel grpcChannel, Properties properties) { + private DaprWorkflowClient(DurableTaskClient innerClient, ManagedChannel grpcChannel) { this.innerClient = innerClient; this.grpcChannel = grpcChannel; - if (properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_ENABLED)) { - this.keepalive = new GrpcChannelKeepalive(grpcChannel, "dapr-workflow-client-keepalive", - properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS)); - } } /** @@ -450,10 +441,6 @@ public boolean purgeWorkflow(String workflowInstanceId) { * Closes the inner DurableTask client and shutdown the GRPC channel. */ public void close() throws InterruptedException { - if (this.keepalive != null) { - this.keepalive.close(); - this.keepalive = null; - } try { if (this.innerClient != null) { this.innerClient.close(); diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java b/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java index e9738e9b01..352164fe34 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java @@ -36,20 +36,33 @@ public final class GrpcChannelKeepalive implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(GrpcChannelKeepalive.class); private static final long KEEPALIVE_DEADLINE_SECONDS = 5; - private final ScheduledExecutorService scheduler; + private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub stub; + private final String threadName; + private final Duration interval; + private ScheduledExecutorService scheduler; /** - * Starts a keepalive loop on the given channel. + * Creates a keepalive for the given channel. No pings are sent until {@link #start()}. * * @param channel channel to the Dapr sidecar to keep alive. * @param threadName name of the keepalive thread, to tell instances apart in thread dumps. * @param interval delay between pings. */ public GrpcChannelKeepalive(Channel channel, String threadName, Duration interval) { - TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub stub = - TaskHubSidecarServiceGrpc.newBlockingStub(channel); - this.scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { - Thread thread = new Thread(runnable, threadName); + this.stub = TaskHubSidecarServiceGrpc.newBlockingStub(channel); + this.threadName = threadName; + this.interval = interval; + } + + /** + * Starts the keepalive loop. Does nothing if already started. + */ + public synchronized void start() { + if (this.scheduler != null) { + return; + } + ScheduledExecutorService newScheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, this.threadName); thread.setDaemon(true); return thread; }); @@ -58,21 +71,25 @@ public GrpcChannelKeepalive(Channel channel, String threadName, Duration interva // every ping after the first fail immediately with DEADLINE_EXCEEDED. // Catch Throwable, not just RuntimeException: any throwable escaping a periodic // task silently cancels all future runs. - this.scheduler.scheduleWithFixedDelay(() -> { + newScheduler.scheduleWithFixedDelay(() -> { try { - stub.withDeadlineAfter(KEEPALIVE_DEADLINE_SECONDS, TimeUnit.SECONDS) + this.stub.withDeadlineAfter(KEEPALIVE_DEADLINE_SECONDS, TimeUnit.SECONDS) .hello(Empty.getDefaultInstance()); } catch (Throwable e) { LOGGER.debug("Sidecar keepalive ping failed", e); } - }, interval.toMillis(), interval.toMillis(), TimeUnit.MILLISECONDS); + }, this.interval.toMillis(), this.interval.toMillis(), TimeUnit.MILLISECONDS); + this.scheduler = newScheduler; } /** - * Stops the keepalive loop. + * Stops the keepalive loop. Does nothing if not started. */ @Override - public void close() { - this.scheduler.shutdownNow(); + public synchronized void close() { + if (this.scheduler != null) { + this.scheduler.shutdownNow(); + this.scheduler = null; + } } } diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java index 1e6742210a..5f7221f6ff 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java @@ -51,8 +51,9 @@ public WorkflowRuntime(DurableTaskGrpcWorker worker, * @param worker grpcWorker processing activities. * @param managedChannel grpc channel. * @param executorService executor service responsible for running the threads. - * @param keepalive application-level keepalive on the worker's channel, stopped on - * {@link #close()}. May be null when no keepalive is wanted. + * @param keepalive application-level keepalive on the worker's channel, started with + * {@link #start()} and stopped on {@link #close()}. May be null when + * no keepalive is wanted. */ public WorkflowRuntime(DurableTaskGrpcWorker worker, ManagedChannel managedChannel, @@ -78,6 +79,9 @@ public void start() { * @param block block the thread if true */ public void start(boolean block) { + if (this.keepalive != null) { + this.keepalive.start(); + } if (block) { this.worker.startAndBlock(); } else { diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java index 371eb01bdb..e812e55aad 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java @@ -86,7 +86,7 @@ public WorkflowRuntime build() { this.executorService = this.executorService == null ? Executors.newCachedThreadPool() : this.executorService; if (instance == null) { GrpcChannelKeepalive keepalive = null; - if (this.properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_ENABLED)) { + if (this.properties.getValue(Properties.WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED)) { keepalive = new GrpcChannelKeepalive(this.managedChannel, "dapr-workflow-runtime-keepalive", this.properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS)); } diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java deleted file mode 100644 index a0283a873f..0000000000 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientKeepaliveTest.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * Copyright 2026 The Dapr Authors - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file 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.dapr.workflows.client; - -import io.dapr.config.Properties; -import org.junit.jupiter.api.Test; - -import java.util.Map; -import java.util.concurrent.TimeUnit; - -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; - -public class DaprWorkflowClientKeepaliveTest { - - private static final String THREAD_NAME = "dapr-workflow-client-keepalive"; - - @Test - public void keepaliveIsOffByDefault() throws InterruptedException { - DaprWorkflowClient client = new DaprWorkflowClient(new Properties()); - try { - assertFalse(keepaliveThreadAlive(), "no keepalive thread expected when the property is off"); - } finally { - client.close(); - } - } - - @Test - public void keepaliveStartsWhenEnabledAndStopsOnClose() throws InterruptedException { - Properties properties = new Properties(Map.of("dapr.workflows.app.keep.alive.enabled", "true")); - DaprWorkflowClient client = new DaprWorkflowClient(properties); - try { - assertTrue(keepaliveThreadAlive(), "keepalive thread expected when the property is on"); - } finally { - client.close(); - } - long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); - while (keepaliveThreadAlive() && System.currentTimeMillis() < deadline) { - Thread.sleep(10); - } - assertFalse(keepaliveThreadAlive(), "keepalive thread should terminate after close()"); - } - - private static boolean keepaliveThreadAlive() { - return Thread.getAllStackTraces().keySet().stream() - .anyMatch(t -> t.getName().equals(THREAD_NAME) && t.isAlive()); - } -} diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java index fa07479f8d..f88a7dbcc0 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java @@ -70,7 +70,7 @@ public WorkflowStub create() { public static void beforeAll() { constructor = Constructor.class.cast(Arrays.stream(DaprWorkflowClient.class.getDeclaredConstructors()) - .filter(c -> c.getParameters().length == 3 + .filter(c -> c.getParameters().length == 2 && c.getParameterTypes()[0] == DurableTaskClient.class) .map(c -> { c.setAccessible(true); @@ -85,7 +85,7 @@ public void setUp() throws Exception { when(mockGrpcChannel.shutdown()).thenReturn(mockGrpcChannel); - client = constructor.newInstance(mockInnerClient, mockGrpcChannel, new Properties()); + client = constructor.newInstance(mockInnerClient, mockGrpcChannel); } @Test diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java index 98ec321439..534d434a7d 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java @@ -72,9 +72,20 @@ public void tearDown() { this.server.shutdownNow(); } + @Test + public void noPingsBeforeStart() throws InterruptedException { + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + Thread.sleep(TEST_INTERVAL.toMillis() * 4); + assertEquals(0, helloCount.get()); + } + } + @Test public void pingsPeriodically() throws InterruptedException { try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + keepalive.start(); + // A second start must not schedule a second ping loop. + keepalive.start(); awaitHelloCountAtLeast(2); } } @@ -83,6 +94,7 @@ public void pingsPeriodically() throws InterruptedException { public void continuesPingingAfterFailures() throws InterruptedException { failPings.set(true); try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + keepalive.start(); awaitHelloCountAtLeast(2); failPings.set(false); awaitHelloCountAtLeast(helloCount.get() + 2); @@ -92,6 +104,7 @@ public void continuesPingingAfterFailures() throws InterruptedException { @Test public void closeStopsPinging() throws InterruptedException { GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL); + keepalive.start(); awaitHelloCountAtLeast(1); keepalive.close(); // Allow an already in-flight ping to finish before snapshotting. @@ -105,6 +118,7 @@ public void closeStopsPinging() throws InterruptedException { public void keepaliveThreadIsDaemonAndStopsOnClose() throws InterruptedException { String threadName = "test-keepalive-lifecycle"; try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, TEST_INTERVAL)) { + keepalive.start(); awaitHelloCountAtLeast(1); Thread thread = findThread(threadName); assertTrue(thread != null && thread.isDaemon(), "expected a live daemon keepalive thread"); diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java index 2eefcfe527..af534b836e 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java @@ -29,6 +29,7 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; public class WorkflowRuntimeTest { @@ -41,12 +42,15 @@ public void startTest() { } @Test - public void closeStopsKeepalive() throws InterruptedException { + public void startStartsKeepaliveAndCloseStopsIt() throws InterruptedException { String threadName = "dapr-workflow-runtime-keepalive"; DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build(); ManagedChannel channel = NetworkUtils.buildGrpcManagedChannel(new Properties()); GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, Duration.ofSeconds(30)); WorkflowRuntime runtime = new WorkflowRuntime(worker, channel, Executors.newCachedThreadPool(), keepalive); + assertFalse(keepaliveThreadAlive(threadName), "keepalive must stay inert until the runtime starts"); + runtime.start(false); + assertTrue(keepaliveThreadAlive(threadName), "keepalive thread expected after runtime start"); assertDoesNotThrow(runtime::close); long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); while (keepaliveThreadAlive(threadName) && System.currentTimeMillis() < deadline) { diff --git a/sdk/src/main/java/io/dapr/config/Properties.java b/sdk/src/main/java/io/dapr/config/Properties.java index 9f5060bac8..eb99fd2671 100644 --- a/sdk/src/main/java/io/dapr/config/Properties.java +++ b/sdk/src/main/java/io/dapr/config/Properties.java @@ -190,21 +190,21 @@ public class Properties { true); /** - * Enables the application-level keepalive on workflow gRPC channels. - * When enabled, the SDK periodically invokes the sidecar's hello RPC so that - * intermediaries that do not treat HTTP/2 PING frames as connection activity + * Enables the application-level keepalive on the workflow runtime's gRPC channel. + * While the runtime is started, the SDK periodically invokes the sidecar's hello RPC + * so that intermediaries that do not treat HTTP/2 PING frames as connection activity * (e.g. AWS ALBs) never see the connection as idle and close it. - * Environment variable: DAPR_WORKFLOWS_APP_KEEP_ALIVE_ENABLED - * System property: dapr.workflows.app.keep.alive.enabled - * Default: false + * Environment variable: DAPR_WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED + * System property: dapr.workflows.runtime.app.keep.alive.enabled + * Default: true */ - public static final Property WORKFLOWS_APP_KEEP_ALIVE_ENABLED = new BooleanProperty( - "dapr.workflows.app.keep.alive.enabled", - "DAPR_WORKFLOWS_APP_KEEP_ALIVE_ENABLED", - false); + public static final Property WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED = new BooleanProperty( + "dapr.workflows.runtime.app.keep.alive.enabled", + "DAPR_WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED", + true); /** - * Interval between application-level keepalive pings on workflow gRPC channels. + * Interval between application-level keepalive pings on the workflow runtime's gRPC channel. * Environment variable: DAPR_WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS * System property: dapr.workflows.app.keep.alive.interval.seconds * Default: 30 seconds