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