KAFKA-16926: Optimize BeginQuorumEpoch heartbeat #20318
Conversation
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
| beginQuorumEpochTimer.update(currentTimeMs); | ||
| for (Map.Entry<ReplicaKey, Long> entry : lastFetchRequestMs.entrySet()) { | ||
| if (beginQuorumEpochTimer.currentTimeMs() - entry.getValue() >= beginQuorumEpochTimeoutMs) { | ||
| replicaKeys.add(ReplicaKey.of(entry.getKey().id(), entry.getKey().directoryId().orElse(NO_DIRECTORY_ID))); |
There was a problem hiding this comment.
how about replicaKeys.add(entry.getKey());?
| .voterKeys() | ||
| .stream() | ||
| .filter(key -> key.id() != quorum.localIdOrThrow()) | ||
| .filter(key -> !needToSendBeginQuorumRequestNode.contains(key.id())) |
There was a problem hiding this comment.
Why are we only checking id? The unique key is actually composed of both id and directory id, right?
There was a problem hiding this comment.
I have updated the code and add a new method equivalentTo to check two replicaKey because the directions can be empty before KIP-853.``
After 4.0 we can assume all node have directoryID, so I rewrite test and use replicaKey.
| private final Timer beginQuorumEpochTimer; | ||
| private final int beginQuorumEpochTimeoutMs; | ||
| private final KafkaRaftMetrics kafkaRaftMetrics; | ||
| private final Map<ReplicaKey, Long> lastFetchRequestMs = new HashMap<>(); |
There was a problem hiding this comment.
This could be replaced with ReplicaState#lastFetchTimestamp, right?
| Set<ReplicaKey> replicaKeys = new HashSet<>(); | ||
| beginQuorumEpochTimer.update(currentTimeMs); | ||
| for (Map.Entry<ReplicaKey, Long> entry : lastFetchRequestMs.entrySet()) { | ||
| if (beginQuorumEpochTimer.currentTimeMs() - entry.getValue() >= beginQuorumEpochTimeoutMs) { |
There was a problem hiding this comment.
From the code above at KafkaRaftClient#L3059,
.filter(key -> !needToSendBeginQuorumRequest.contains(key))I think the condition should be changed to:
| if (beginQuorumEpochTimer.currentTimeMs() - entry.getValue() >= beginQuorumEpochTimeoutMs) { | |
| if (beginQuorumEpochTimer.currentTimeMs() - entry.getValue() < beginQuorumEpochTimeoutMs) { |
and we should rename the method to skipSendBeginQuorumRequestNodes.
There was a problem hiding this comment.
I also use needToSendBeginQuorumRequest and change filter condition because positive logic is easier to understand.
TaiJuWu
force-pushed
the
KAFKA-16926
branch
from
53eb81c to
f67e5bd
Compare
Instead of sending out BeginQuorum requests to every voter on a cadence,
we can save on some requests by only sending to those which have not
fetched within the BeginQuorumEpoch timeout.