Add worker thread #1

siri-varma · 2025-05-07T06:02:01Z

Issue describing the changes in this PR

resolves #
github.com/dapr/java-sdk/issues/1331

Pull request checklist

My changes do not require documentation changes
- Otherwise: Documentation issue linked to PR
My changes are added to the CHANGELOG.md
I have added all required tests (Unit tests, E2E tests)

Additional information

Additional PR information

siri-varma · 2025-05-07T06:30:16Z

@yaron2 , @cgillum I plan to test this PR out tomorrow. Can you folks tell me your initial thoughts on this please ? This is the github issue: github.com/dapr/java-sdk/issues/1331

siri-varma · 2025-05-07T06:32:27Z

I looked at dotnet durable task implementation. It just uses a simple run https://github.com/microsoft/durabletask-dotnet/blob/main/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs#L333

So followed a similar implementation

rishi-dev89 · 2025-05-07T10:03:05Z

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

        this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel);
        this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter();
        this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval : DEFAULT_MAXIMUM_TIMER_INTERVAL;
+        this.workerPool = Executors.newCachedThreadPool();


Please allow client to pass the workerPool, so that client can make use of the VirtualThread pool if and when required.
2. Await for the executor to shutdown before closing the connection to mitigate the risk on inflight requests being processed by threadpool

For the second point, we need to use the this.workerPoolshutdown() method in the public void stop() {} function.

Provided the ability to pass in the worker poll too and added the workerPoolShutdown

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

salaboy · 2025-05-07T15:37:12Z

@siri-varma @rishi-dev89 ok.. I am looking at this PR and I am trying to understand a bit the expected behaviour that is implemented.
If we use a thread pool then tasks will be executed as scheduled, so the behavior might not be the same if we have concurrent tasks. If we are ok with that change in behaviour this PR looks like ok to me, but I might be lacking some context.

salaboy · 2025-05-07T15:38:15Z

Also the TODO validated that this was the original intention from the author:
//TODO: Run this on a worker pool thread: https://www.baeldung.com/thread-pool-java-and-guava

siri-varma · 2025-05-07T16:09:48Z

Validated that it is running on different threads

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

Signed-off-by: siri-varma <[email protected]>

cgillum

Overall LGTM. I especially appreciate the change to allow the user code to provide the executor pool. Just one main concern about the management of the worker's while loop.

cgillum · 2025-05-07T17:19:57Z

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java


        // TODO: How do we interrupt manually?
-        while (true) {
+        while (!this.workerPool.isShutdown()) {


I'm wondering what the side effects of this change might be. For sure we'll need to update the logging code to account for this new exit condition. But does this mean that user code could also accidentally shut down this loop by shutting down the worker pool? Is this change critical?

Once the workerpool is closed they cannot submit new activities anyway so I thought we could safely come out of the while loop.

If that's the case, we should at least log a warning.

is this resolved? @siri-varma

Yes. Took care of it.

siri-varma · 2025-05-07T17:32:23Z

@cgillum, @JoshVanL, @rishi-dev89, @cicoyle, @salaboy Addressed all the feedback on this PR. Thank you for the reviews

Signed-off-by: siri-varma <[email protected]>

rishi-dev89 · 2025-05-07T17:49:05Z

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

+                                    .setCustomStatus(StringValue.of(taskOrchestratorResult.getCustomStatus()))
+                                    .build();
+
+                            this.sidecarClient.completeOrchestratorTask(response);


Please use a try catch here as well as it can throw StatusRuntimeException

catch (StatusRuntimeException e) {
if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
logger.log(Level.INFO, "The sidecar at address {0} is unavailable. Will continue retrying.", this.getSidecarAddress());
} else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
logger.log(Level.INFO, "Durable Task worker has disconnected from {0}.", this.getSidecarAddress());
logger.log(Level.INFO, "Durable Task worker has disconnected from {0}.", this.getSidecarAddress());
} else {
logger.log(Level.WARNING, "Unexpected failure connecting to {0}.", this.getSidecarAddress());
}

rishi-dev89 · 2025-05-07T17:50:15Z

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

+                                responseBuilder.setFailureDetails(failureDetails);
+                            }
+
+                            this.sidecarClient.completeActivityTask(responseBuilder.build());


Try catch block would be required here as well, otherwise the exception will never be logged and could be difficult to debbug

good idea. I will do a quick small PR to address this

cgillum · 2025-05-07T17:59:06Z

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

    }

    private void shutDownWorkerPool() {
+        logger.log(Level.WARNING, "ExecutorService shutdown initiated. No new tasks will be accepted");


This is not correct. Now this warning will be logged every time a user shuts down their worker, but we only want to log the warning when the worker pool was shut down unexpectedly.

I opened a PR here. I think this addresses your comment.

Fix release package name

rishi-dev89 reviewed May 7, 2025

View reviewed changes

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java Show resolved Hide resolved

siri-varma marked this pull request as ready for review May 7, 2025 16:06

JoshVanL suggested changes May 7, 2025

View reviewed changes

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java Outdated Show resolved Hide resolved

siri-varma added 3 commits May 7, 2025 09:20

Add worker thread

43e11e5

Signed-off-by: siri-varma <[email protected]>

Add worker thread

9d4bae0

Signed-off-by: siri-varma <[email protected]>

Add worker thread

5b2db59

Signed-off-by: siri-varma <[email protected]>

siri-varma force-pushed the users/svegiraju/add-thread branch from 7ecfa96 to 5b2db59 Compare May 7, 2025 16:20

Add check

a051650

Signed-off-by: siri-varma <[email protected]>

cgillum reviewed May 7, 2025

View reviewed changes

JoshVanL approved these changes May 7, 2025

View reviewed changes

cicoyle approved these changes May 7, 2025

View reviewed changes

Added warn

faf6ff7

Signed-off-by: siri-varma <[email protected]>

cicoyle merged commit fb9c30d into dapr:main May 7, 2025
1 check passed

rishi-dev89 reviewed May 7, 2025

View reviewed changes

cgillum reviewed May 7, 2025

View reviewed changes

cicoyle mentioned this pull request May 7, 2025

try-catch #4

Merged

JoshVanL pushed a commit to JoshVanL/durabletask-java that referenced this pull request May 7, 2025

Merge pull request dapr#1 from JoshVanL/fix-release-package

d3ce5c9

Fix release package name

siri-varma deleted the users/svegiraju/add-thread branch May 7, 2025 23:17

cicoyle mentioned this pull request May 8, 2025

Workflow Activity code is directly run via DurableTaskGrpcWorker dapr/java-sdk#1331

Closed

Add worker thread #1

Add worker thread #1

Uh oh!

Conversation

siri-varma commented May 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Issue describing the changes in this PR

Pull request checklist

Additional information

Uh oh!

siri-varma commented May 7, 2025

Uh oh!

siri-varma commented May 7, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

salaboy commented May 7, 2025

Uh oh!

salaboy commented May 7, 2025

Uh oh!

siri-varma commented May 7, 2025

Uh oh!

Uh oh!

cgillum left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

siri-varma May 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

siri-varma commented May 7, 2025

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

siri-varma commented May 7, 2025 •

edited

Loading

siri-varma May 7, 2025 •

edited

Loading