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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
