KAFKA-19826: Implement coordinator adaptive batch linger time #20780
Conversation
NB: This refers to the CoordinatorRuntimeBuilder's default append linger time, not the config-level default.
github-actions
bot
added
triage
| private final Deque<CoordinatorEvent> queue; | ||
| private boolean inEvent; | ||
|
|
||
| public DirectEventProcessor() { | ||
| this.queue = new LinkedList<>(); | ||
| this.inEvent = false; | ||
| } | ||
|
|
There was a problem hiding this comment.
We can now enqueue flush events while in the middle of another event. The flush event must be run after the current event, so we have to introduce a queue here.
| .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class)) | ||
| .withCoordinatorMetrics(mock(CoordinatorMetrics.class)) | ||
| .withSerializer(new StringSerializer()) | ||
| .withAppendLingerMs(0) |
There was a problem hiding this comment.
Many of the tests in this file run with an adaptive linger time now, except for those with an explicit linger time.
I updated a handful of tests using the ManualEventProcessor to use an append linger time of 0 (preserving the existing behavior) to avoid test churn.
|
The changes are a little long, so it may be easier to review this PR commit by commit, at least for the first 5 commits. |
| public Builder<S, U> withAppendLingerMs(int appendLingerMs) { | ||
| this.appendLingerMs = OptionalInt.of(appendLingerMs); | ||
| return this; | ||
| } |
There was a problem hiding this comment.
From the references, this function is only used by test code. Should we remove it and update test cases to use OptionalInt one?
There was a problem hiding this comment.
Sure, we can remove the method.
| if (appendLingerMs == null) | ||
| appendLingerMs = OptionalInt.empty(); |
There was a problem hiding this comment.
From the production code, the appendLingerMs input cannot be null. How about we set a default value OptionalInt.empty() to appendLingerMs and remove appendLingerMs == null check?
There was a problem hiding this comment.
I was following the existing pattern in the builder. We don't set a default value for logPrefix, logContext and compression and initialize them in build().
| if (appendLingerMs.isPresent() && appendLingerMs.getAsInt() < -1) | ||
| throw new IllegalArgumentException("AppendLinger must be empty or >= 0"); |
There was a problem hiding this comment.
If the appendLingerMs is -1, it will be input as OptionalInt.empty(). How about checking the value is >= 0 here?
There was a problem hiding this comment.
Good catch! The check is supposed to reject -1.
|
@FrankYang0529 Thanks for the review! I updated the code. I also added some tests for the config logic. |
3 participants
Add support for an adaptive batch linger time in the group and share
coordinators. When an adaptive batch linger time is enabled, we no
longer create a timer to flush the current batch. Instead, we append a
flush operation at the end of the event queue so that any currently
queued operations are naturally collected into the batch.
To avoid double flushing from hitting the maximum batch size or
transactional writes, we number batches with an epoch to check whether
the batch has already been flushed.
The group.coordinator.append.linger.ms and
share.coordinator.append.linger.ms configs are extended to allow -1, to
specify an adaptive append linger time. The default for these configs is
also updated to -1.