KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) #17720
Conversation
github-actions
bot
added
streams
core
There was a problem hiding this comment.
Thanks for the PR @rajinisivaram. Just some questions for my understanding.
| private void rebootstrap(long now) { | ||
| closeAll(); | ||
| metadata.rebootstrap(); | ||
| metadataAttemptStartMs = Optional.of(now); |
There was a problem hiding this comment.
Question: Do we need to set the time while rebootstrapping or reset to Optional.empty? As whenever there would be successful response from metadata then the time will be updated?
Or is it more to handle that after re-bootstrap, if we never receive successful metadata response then we would like to rebootstrap again?
There was a problem hiding this comment.
It is to handle rebootstrapping again if successful metadata response is not received after one rebootstrap (the second case you mentioned).
|
|
||
| if (!cluster.nodes().isEmpty()) { | ||
| this.cluster = cluster; | ||
| } | ||
| } | ||
|
|
||
| public void initiateRebootstrap() { | ||
| requestUpdate(); | ||
| this.metadataAttemptStartMs = Optional.of(0L); |
There was a problem hiding this comment.
Question: Why don't we set the time to now as we do in NetworkClient?
There was a problem hiding this comment.
In rebootstrap(), we set time to now as in NetworkClient. In both, when we receive REBOOTSTRAP_REQUIRED error code, we set to 0, so that rebootstrap is triggered on the next poll regardless of time.
| logContext, | ||
| Long.MAX_VALUE, |
There was a problem hiding this comment.
Seems this constructor is used be Connect Worker which gets passed to ConsumerNetworkClient. So should we not support the consumer client spun in other dependant modules? Do you think the a default value of Long.MAX_VALUE for this config makes sense so we can just get the ocnfigured vs default value?
There was a problem hiding this comment.
Did you mean the NetworkClient in WorkerGroupMember? This PR changes the constructor used in WorkerGroupMember to pass in the configured value. Let me know if I have missed a different one.
|
@apoorvmittal10 Thanks for the review, I have responded to the comments. |
rajinisivaram
force-pushed
the
KAFKA-17885-rebootstrap
branch
from
38bc9dd
to
1759594
Compare
Implementation of https://cwiki.apache.org/confluence/display/KAFKA/KIP-1102%3A+Enable+clients+to+rebootstrap+based+on+timeout+or+error+code
metadata.recovery.rebootstrap.trigger.ms, set to 5 minutes by defaultrebootstrapthe default formetadata.recovery.strategyREBOOTSTRAP_REQUIRED, introduces top-level error code in metadata response. On this error, clients rebootstrap.Committer Checklist (excluded from commit message)