You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chickenchickenlove (via GitHub)" <gi...@apache.org> on 2026/10/22 12:53:55 UTC
[PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
chickenchickenlove opened a new pull request, #23913:
URL: https://github.com/apache/kafka/pull/23913
This PoC makes `CoordinatorTimerImpl` snapshottable, aligning timer state restoration with the Coordinator’s snapshot and rollback lifecycle.
The main changes are:
1. `CoordinatorTimerImpl` stores a `TimerState` for each timer in a `TimelineHashMap`, using the same `SnapshotRegistry` as the Coordinator.
2. `TimerState` contains the information needed to recreate a `TimerTask`: its deadline, operation, and retry configuration. Scheduled `TimerTask` instances are managed separately.
3. When a record append succeeds, the existing `updateLastWrittenOffset(offset)` flow creates a snapshot at that offset, including the timer state. Scheduled tasks continue running unchanged.
4. If the append fails, `revertLastWrittenOffset()` restores the timer state to the rollback target snapshot.
5. After restoring the state, `timer.onRollback()` cancels all currently tracked tasks and schedules new tasks based on the restored `TimerState entries`. Each task uses the time remaining until its original deadline, or a delay of zero if the deadline has already passed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4351867990
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@chickenchickenlove @dajac I tried the approach for the coordinator executor in https://github.com/apache/kafka/pull/23923. Names of the commit/revert methods are tbd and we should align on them. I think the requirements for the executor are a lot more relaxed than timers - we only need to cancel tasks on a rollback and can rely on the next heartbeat to re-schedule tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23913:
URL: https://github.com/apache/kafka/pull/23913#issuecomment-6341357308
Hi, @squah-confluent !
This is a draft PoC PR for the snapshottable timer.
It is still at an early stage(not cover all details), but I think the basic implementation is mostly in place.
Could we discuss the implementation here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4359239838
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@dajac
Agree. This implementation is expensive and not scalable because of time complexity. (@squah-confluent also, thanks for taking time to look into this!)
I've given some thought to the approach based on your suggestions. What do you think of this implementation?
1. Timer `A` is scheduled with a deadline of 20. There is no pending timer.
2. The coordinator schedules a replacement, Timer `B`, with the same key and a deadline of 40. `A` remains scheduled, while `B` is kept pending.
3. Depending on the write result:
- Success: The coordinator runtime calls `commit()`, cancels `A`, and schedules `B` using the remaining time until its deadline. The pending entry is cleared.
- Failure: The coordinator runtime calls `revert()` and discards `B`. `A` remains scheduled.
There are also two cases we need to handle these
1. Operations that produce no records. If `commit()` is only called after a successful log write, their pending timer changes may never be applied. We therefore need a way to finalize these changes without a new write, while respecting any pending batch they depend on.
2. A expires while `B` is pending. We could retain `A`’s timeout operation without executing it. On `commit()`, we discard that operation and schedule `B`. On `revert()`, we discard `B` and enqueue `A`’s timeout operation for execution after the coordinator state has been rolled back.
What do you think?
@dajac @squah-confluent When you get a chance, please take a look!! 🙇♂️
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4378971691
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
> 2. Timers scheduled by failure callbacks to restore the previous state must survive that recovery.
I'm expecting these to be a manual implementation of rolling back timer state. Hopefully they should all disappear at the end of this PR?
> 3. Changes such as extending a session timeout upon receiving a heartbeat require a separate policy decision: should they also be reverted when an unrelated write fails?
For simplicity I think unextending the session timeout is best. A consumer would have to miss more than one heartbeat to be evicted from a group. We could consider write failures to be rare enough that it's not worth the extra design effort. @dajac any thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4359286384
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
> 1. Operations that produce no records.
Not sure if this is 100% a match for the case you have in mind: We can "commit" instantly in `append()` when there are no records to write and there is no pending batch. (I'm still stuck on the names.)
https://github.com/apache/kafka/pull/23923/changes#diff-37d422ba9671aaf4524a78e25a9c50830ca0aaf64e1979772c70e4e1198c919bR925
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on PR #23913:
URL: https://github.com/apache/kafka/pull/23913#issuecomment-6341656727
Thanks! I'll take a look when I can!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4364992365
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
Agreed! I also like the method names you suggested.
I'm going to work on implementing the 2 phase approach on the `CoordinatorTimerImpl` side.
This PR still uses TimelineHashMap for the snapshottable implementation. 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4351867990
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@chickenchickenlove @dajac I tried the approach for the coordinator executor in https://github.com/apache/kafka/pull/23923. The names of the commit/revert methods are tbd we should align on them. I think the requirements for the executor are a lot more relaxed than timers - we only need to cancel tasks on a rollback and can rely on the next heartbeat to re-schedule tasks.
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@chickenchickenlove @dajac I tried the approach for the coordinator executor in https://github.com/apache/kafka/pull/23923. The names of the commit/revert methods are tbd we should align on them. The requirements for the executor look a lot more relaxed than timers - we only need to cancel tasks on a rollback and can rely on the next heartbeat to re-schedule tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on PR #23913:
URL: https://github.com/apache/kafka/pull/23913#issuecomment-6345658764
I have not fully reviewed yet. Can we make sure we think about the case where a timer fires, tries to write records but persistently fails?
eg. the last consumer protocol member expiring from a mixed group, whose departure triggers a downgrade to a classic group and a write of a large `GroupMetadata` record. If the write is oversized and always fails, we can get stuck in a loop if we roll back the timer and re-run it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4359239838
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@dajac
Agree. This implementation is expensive and not scalable because of time complexity.
I've given some thought to the approach based on your suggestions. What do you think of this implementation?
1. Timer `A` is scheduled with a deadline of 20. There is no pending timer.
2. The coordinator schedules a replacement, Timer `B`, with the same key and a deadline of 40. `A` remains scheduled, while `B` is kept pending.
3. Depending on the write result:
- Success: The coordinator runtime calls `commit()`, cancels `A`, and schedules `B` using the remaining time until its deadline. The pending entry is cleared.
- Failure: The coordinator runtime calls `revert()` and discards `B`. `A` remains scheduled.
There are also two cases we need to handle these
1. Operations that produce no records. If `commit()` is only called after a successful log write, their pending timer changes may never be applied. We therefore need a way to finalize these changes without a new write, while respecting any pending batch they depend on.
2. A expires while `B` is pending. We could retain `A`’s timeout operation without executing it. On `commit()`, we discard that operation and schedule `B`. On `revert()`, we discard `B` and enqueue `A`’s timeout operation for execution after the coordinator state has been rolled back.
What do you think?
@dajac @squah-confluent When you get a chance, please take a look!! 🙇♂️
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4369854996
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@dajac @squah-confluent
I tried implementing an approach where timer changes remain pending until the batch is successfully appended to the local log. On success, the changes are applied; on failure, they are discarded.
This seems suitable for issues such as KAFKA-21279, where timer changes need to be reverted when a write fails. However, applying the same policy to all existing `schedule()` and `cancel()` calls revealed that these changes have different semantics:
1. Changes that depend on speculative coordinator state should be discarded if the corresponding write fails.
2. Timers scheduled by failure callbacks to restore the previous state must survive that recovery.
3. Changes such as extending a session timeout upon receiving a heartbeat require a separate policy decision: should they also be reverted when an unrelated write fails?
The current `schedule()` and `cancel()` APIs do not express these dependencies. Having the Runtime or TimerImpl infer them from the execution path makes the implementation more complex and risks unintentionally changing existing behavior.
What do you think about preserving the existing APIs’ immediate scheduling and cancellation behavior, while allowing callers to explicitly choose when timer changes should depend on the write outcome?
For example, `CoordinatorRuntime` could own both the existing `CoordinatorTimerImpl` and a separate `RevertableCoordinatorTimerImpl`, and expose both to the coordinator. Callers would use the existing timer for changes that should take effect immediately, and explicitly use the revertable timer for changes that should be applied or discarded based on the write outcome.
If both implementations can operate on timers with the same key, we would also need to define how replacement and cancellation interact across them, including the order in which changes are applied.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement snapshottable timers in the coordinator runtime [kafka]
Posted by "chickenchickenlove (via GitHub) via jira" <ji...@kafka.apache.org>.
chickenchickenlove commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4434569483
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@squah-confluent
Thanks, that makes sense. I agree that timers scheduled solely to restore previous timer state should no longer need manual recovery once we have common handling for this. I’m also open to reverting heartbeat timeout extensions for simplicity, as you suggested.
My remaining concern is that some timers scheduled by failure callbacks represent new recovery actions rather than restoration of previous timer state. For example, when persisting a classic group’s assignment fails during SyncGroup, the failure callback can initiate a new rebalance and schedule a join timeout. That timer would still be needed after the failed write is rolled back.
This makes me think we need to distinguish changes that belong to the failed write from changes that should survive it, while also associating write-dependent changes with the correct batch.
My understanding is that the core goal of this ticket is to keep timers consistent with the coordinator state after a rollback. For example, if a local log append fails and a member removal is rolled back, the cancellation of that member’s session timer should also be reverted.
I agree that code which manually restores previous timer state should be replaced by common handling. However, not all timer changes depend on record writes in the same way. For example, a timer scheduled to start a new recovery action after a failure needs to be distinguished from changes belonging to the failed write.
Also, operation boundaries do not always align with batch boundaries:
1. Even an atomic operation may change timers while generating its records, then flush an existing batch before appending its own records.
2. A non-atomic operation may have its records split across multiple batches, with only a prefix successfully written.
Therefore, I don’t think it is sufficient to apply all pending timer changes whenever a batch succeeds, or discard them all whenever an operation fails. We need to associate each timer change with the write it actually depends on.
Would the following approach make sense?
1. Keep the existing immediate scheduling and cancellation behavior for timer changes that are independent of record writes.
2. Explicitly associate write-dependent timer changes with the operation generating the records, and keep those changes pending.
3. Have CoordinatorRuntime attach those changes to the batch that actually contains the corresponding records, and invoke common success/failure handling based on the local append outcome.
These callbacks would need to be distinct from the existing appendFuture, whose completion also depends on the high watermark and can fail due to an operation timeout. For non-atomic operations, a single final callback cannot represent partial success, so we would need either explicit record dependency boundaries or a restriction on the initial scope of support.
This would let callers explicitly declare write dependencies while keeping the application, discarding, and expiration handling of pending timer changes in a common layer.
What do you think, @squah-confluent @dajac ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4351867990
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@chickenchickenlove @dajac I tried the batch-based approach for the coordinator executor in https://github.com/apache/kafka/pull/23923. The names of the commit/revert methods are tbd we should align on them. The requirements for the executor look a lot more relaxed than timers - we only need to cancel tasks on a rollback and can rely on the next heartbeat to re-schedule tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19023: Implement Snapshottable timers in the coordinator runtime. [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #23913:
URL: https://github.com/apache/kafka/pull/23913#discussion_r4349756755
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java:
##########
@@ -166,6 +225,30 @@ public boolean isScheduled(String key) {
}
public void cancelAll() {
+ states.clear();
+ cancelTasks();
+ }
+
+ /**
+ * Rebuilds the scheduled tasks after the supplied snapshot registry has been reverted.
+ * This must be called under the same external synchronization as the registry rollback.
+ * It preserves the restored states and their original deadlines.
+ */
+ public void onRollback() {
+ cancelTasks();
+ // A timer implementation may execute an expired task immediately, so iterate over
+ // a copy in case that task updates the states while we are rescheduling them.
+ new HashMap<>(states).forEach(
+ (key, state) ->
+ scheduleTask(
+ key,
+ state,
+ Math.max(0, state.deadlineMs() - time.hiResClockMs())
+ )
+ );
+ }
Review Comment:
@chickenchickenlove Thanks for the patch! Does it mean that we must cancel and re-schedule all the timers on rollbacks? This seems pretty expensive, isn't it?
I'd like to suggest an alternative design. One approach would be to only keep the timers made by the operations of the current batch as pending until the batch is successfully written. When it is, the pending timers are moved to the "committed" space. If the write fails, they are dropped. We don't really need to full-fleshed rollback to any previous committed offset in the timer case.
It basically means that a timer is not really overridden until committed. In this case, we must define what to do if the previous one fires before the new one is committed. This may also be true in your design.
I have not tested it. It is just an idea. Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org