MINOR: Decouple timer and executor from CoordinatorRuntime #21350

dajac · 2026-01-23T17:39:48Z

This patch refactors the coordinator runtime to improve modularity by
decoupling the timer and executor components:

Extract EventBasedCoordinatorTimer from CoordinatorRuntime into
standalone CoordinatorTimerImpl;
Introduce CoordinatorShardScheduler interface to allow shard-scoped
components (timer, executor) to schedule write operations back to the
runtime within their shard's scope;
Simplify CoordinatorExecutorImpl by replacing direct runtime
dependency with the new scheduler interface;

Reviewers: Sean Quah squah@confluent.io, Chia-Ping Tsai
chia7712@gmail.com

- Replace CoordinatorRuntime and TopicPartition dependencies with a functional Scheduler interface - Remove unnecessary generic type parameter S - Use var for local variable declarations - Simplify tests by mocking the Scheduler instead of CoordinatorRuntime

This patch extracts EventBasedCoordinatorTimer from CoordinatorRuntime into a standalone CoordinatorTimerImpl class, following the same pattern as CoordinatorExecutorImpl. The new class uses a Scheduler functional interface to decouple from CoordinatorRuntime internals. This also adds comprehensive unit tests for CoordinatorTimerImpl.

Extract the identical WriteOperation and Scheduler interfaces from CoordinatorExecutorImpl and CoordinatorTimerImpl into a shared CoordinatorShardScheduler interface. This provides a common internal API for shard-scoped components to schedule write operations through the coordinator runtime.

Reorder constructor parameters to place the scheduler last, matching the parameter order in CoordinatorTimerImpl.

dajac · 2026-01-23T20:18:21Z

@squah-confluent Could you please review?

chia7712

@dajac thanks for this great refactor

chia7712 · 2026-01-23T22:43:16Z

...r-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java

+    @Override
+    public void schedule(
+        String key,
+        long delay,


Have you considered using a Duration to replace both delay and unit?

I have not considered it. It will consider it separately and raise a PR if it makes sense to do it.

chia7712 · 2026-01-23T22:57:12Z

...r-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java

+        };
+
+        log.debug("Registering timer {} with delay of {}ms.", key, unit.toMillis(delay));
+        var prevTask = tasks.put(key, task);


var prevTask = tasks.put(key, task);
if (!tasks.remove(key, this))

Is it possible that two different threads change the tasks concurrently after this decoupling?

No, the threading model is still the same.

the threading model is still the same.

If scheduler.scheduleWriteOperation returns a failed CompletableFuture, the exceptionally block will be executed by the timer thread. Does it break the threading model?

chia7712 · 2026-01-23T22:59:26Z

...r-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java

+                        // Execute the timeout operation.
+                        return operation.generateRecords();
+                    }
+                ).exceptionally(ex -> {


What if scheduleWriteOperation returns a failed future? Will the task not be removed?

Good catch. We should remove it here.

squah-confluent · 2026-01-24T10:40:13Z

...tor-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

-                    log.debug("Scheduling write event {} for timer {}.", event.name, key);
-                    try {
-                        enqueueLast(event);
-                    } catch (NotCoordinatorException ex) {


The exceptions that come out of scheduleWriteOperation/enqueueLast are very confusing. I think it would help if we were a bit more thorough with the @throwss in the javadocs, not necessarily in this PR.

~~I think the RejectedExecutionException handler within the exceptionally above will never be hit because scheduleWriteOperation will throw it directly?~~ seems that both the future can complete with RejectedExecutionException and scheduleWriteOperation can throw it directly.

This patch fixes two issues in CoordinatorTimerImpl: 1. Synchronous exceptions from scheduleWriteOperation were not caught, leaving tasks in the map. Fixed by wrapping the scheduler lambda with try-catch in CoordinatorRuntime.CoordinatorContext to convert exceptions to failed futures. 2. Tasks were not removed in the exceptionally handler when the write operation failed. Fixed by adding defensive task cleanup as the first step in the exceptionally handler, following the same pattern as CoordinatorExecutorImpl.

dajac · 2026-01-24T10:52:41Z

The exceptions that come out of scheduleWriteOperation/enqueueLast are very confusing. I think it would help if we were a bit more thorough with the @throwss in the javadocs, not necessarily in this PR.

Yes, I agree.

I think the RejectedExecutionException handler within the exceptionally above will never be hit because scheduleWriteOperation will throw it directly? seems that both the future can complete with RejectedExecutionException and scheduleWriteOperation can throw it directly.

scheduleWriteOperation should not throw RejectedExecutionException but the timer and the executor use it internally. The part which annoys me is that scheduleWriteOperation can throw synchronously and fail the future. It would be better to only use one pattern. I will try to improve this separately too.

dajac · 2026-01-24T10:54:06Z

...tor-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

+                    try {
+                        return scheduleWriteOperation(
+                            operationName,
+                            tp,
+                            coordinator -> operation.generate()
+                        );
+                    } catch (Throwable t) {
+                        return CompletableFuture.failedFuture(t);
+                    }


I handle synchronous exceptions here so components can only work with the future. It is simpler this way.

squah-confluent

Thanks for the refactor!

chia7712 · 2026-01-26T08:27:42Z

scheduleWriteOperation should not throw RejectedExecutionException but the timer and the executor use it internally. The part which annoys me is that scheduleWriteOperation can throw synchronously and fail the future. It would be better to only use one pattern. I will try to improve this separately too.

agreed. It would be cool to handle both in one fell swoop. The method returning CompletableFuture should NOT throw exception directly

dajac added 4 commits January 23, 2026 18:29

MINOR: Align constructor parameter order in CoordinatorExecutorImpl

38e14bc

Reorder constructor parameters to place the scheduler last, matching the parameter order in CoordinatorTimerImpl.

github-actions bot added the group-coordinator label Jan 23, 2026

dajac requested a review from lianetm January 23, 2026 20:17

chia7712 reviewed Jan 23, 2026

View reviewed changes

squah-confluent reviewed Jan 24, 2026

View reviewed changes

dajac commented Jan 24, 2026

View reviewed changes

squah-confluent approved these changes Jan 24, 2026

View reviewed changes

dajac requested a review from chia7712 January 24, 2026 11:57

chia7712 approved these changes Jan 26, 2026

View reviewed changes

chia7712 merged commit 5f0e368 into apache:trunk Jan 26, 2026
26 checks passed

dajac deleted the minor-timer-cleanup branch January 26, 2026 09:15

MINOR: Decouple timer and executor from CoordinatorRuntime #21350

MINOR: Decouple timer and executor from CoordinatorRuntime #21350

Conversation

dajac commented Jan 23, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dajac commented Jan 23, 2026

Uh oh!

chia7712 left a comment

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!

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!

dajac commented Jan 24, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

squah-confluent left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Jan 26, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dajac commented Jan 23, 2026 •

edited by github-actions bot

Loading