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