You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "rajinisivaram (via GitHub)" <gi...@apache.org> on 2024/11/08 12:17:12 UTC
[PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
rajinisivaram opened a new pull request, #17720:
URL: https://github.com/apache/kafka/pull/17720
Implementation of https://cwiki.apache.org/confluence/display/KAFKA/KIP-1102%3A+Enable+clients+to+rebootstrap+based+on+timeout+or+error+code
- Introduces rebootstrap trigger interval config `metadata.recovery.rebootstrap.trigger.ms`, set to 5 minutes by default
- Makes `rebootstrap` the default for `metadata.recovery.strategy`
- Adds new error code `REBOOTSTRAP_REQUIRED`, introduces top-level error code in metadata response. On this error, clients rebootstrap.
- Configs apply to producers, consumers, share consumers, admin clients, Connect and KStreams clients.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1840224877
##########
core/src/test/scala/integration/kafka/api/ConsumerRebootstrapTest.scala:
##########
@@ -87,3 +128,17 @@ class ConsumerRebootstrapTest extends RebootstrapTest {
producer.close()
}
}
+
+object ConsumerRebootstrapTest {
+
+ final val RebootstrapTestName = s"${TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames}.useRebootstrapTriggerMs={2}"
+ def rebootstrapTestParams: stream.Stream[Arguments] = {
+ assertEquals(1, getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_ZK_implicit.count())
Review Comment:
@omkreddy Thanks for the review. The code below is using only the first entry. At the moment, there is only one entry, but I wanted to make sure that if we added more in future, the test would get updated to include all.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1834643738
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##########
@@ -289,17 +304,24 @@ public void update(Cluster cluster, long now) {
this.state = State.QUIESCENT;
this.fatalException = null;
+ this.metadataAttemptStartMs = Optional.empty();
if (!cluster.nodes().isEmpty()) {
this.cluster = cluster;
}
}
+ public void initiateRebootstrap() {
+ requestUpdate();
+ this.metadataAttemptStartMs = Optional.of(0L);
Review Comment:
Question: Why don't we set the time to `now` as we do in NetworkClient?
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1236,6 +1316,12 @@ public void close() {
this.metadata.close();
}
+ private void rebootstrap(long now) {
+ closeAll();
+ metadata.rebootstrap();
+ metadataAttemptStartMs = Optional.of(now);
Review 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?
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -166,11 +169,51 @@ public NetworkClient(Selectable selector,
time,
discoverBrokerVersions,
apiVersions,
- null,
logContext,
+ Long.MAX_VALUE,
Review 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?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on PR #17720:
URL: https://github.com/apache/kafka/pull/17720#issuecomment-2466164414
@apoorvmittal10 Thanks for the review, I have responded to the comments.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1879929515
##########
core/src/test/scala/integration/kafka/api/ConsumerRebootstrapTest.scala:
##########
@@ -87,3 +128,17 @@ class ConsumerRebootstrapTest extends RebootstrapTest {
producer.close()
}
}
+
+object ConsumerRebootstrapTest {
+
+ final val RebootstrapTestName = s"${TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames}.useRebootstrapTriggerMs={2}"
+ def rebootstrapTestParams: stream.Stream[Arguments] = {
+ assertEquals(1, getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_ZK_implicit.count())
Review Comment:
Why did we add new tests depending on ZooKeeper in trunk? We're in the process of removing ZooKeeper.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1835335426
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##########
@@ -289,17 +304,24 @@ public void update(Cluster cluster, long now) {
this.state = State.QUIESCENT;
this.fatalException = null;
+ this.metadataAttemptStartMs = Optional.empty();
if (!cluster.nodes().isEmpty()) {
this.cluster = cluster;
}
}
+ public void initiateRebootstrap() {
+ requestUpdate();
+ this.metadataAttemptStartMs = Optional.of(0L);
Review 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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on PR #17720:
URL: https://github.com/apache/kafka/pull/17720#issuecomment-2473551972
@apoorvmittal10 @omkreddy Thanks for the reviews. @apoorvmittal10 If you have follow-on comments, I can address those in a separate PR. Merging this to trunk.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1835335844
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -1236,6 +1316,12 @@ public void close() {
this.metadata.close();
}
+ private void rebootstrap(long now) {
+ closeAll();
+ metadata.rebootstrap();
+ metadataAttemptStartMs = Optional.of(now);
Review Comment:
It is to handle rebootstrapping again if successful metadata response is not received after one rebootstrap (the second case you mentioned).
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1835335638
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -166,11 +169,51 @@ public NetworkClient(Selectable selector,
time,
discoverBrokerVersions,
apiVersions,
- null,
logContext,
+ Long.MAX_VALUE,
Review 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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #17720:
URL: https://github.com/apache/kafka/pull/17720#discussion_r1839870397
##########
core/src/test/scala/integration/kafka/api/ConsumerRebootstrapTest.scala:
##########
@@ -87,3 +128,17 @@ class ConsumerRebootstrapTest extends RebootstrapTest {
producer.close()
}
}
+
+object ConsumerRebootstrapTest {
+
+ final val RebootstrapTestName = s"${TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames}.useRebootstrapTriggerMs={2}"
+ def rebootstrapTestParams: stream.Stream[Arguments] = {
+ assertEquals(1, getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_ZK_implicit.count())
Review Comment:
nit: why do we need this assert statement?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17885: Enable clients to rebootstrap based on timeout or error code (KIP-1102) [kafka]
Posted by "rajinisivaram (via GitHub)" <gi...@apache.org>.
rajinisivaram merged PR #17720:
URL: https://github.com/apache/kafka/pull/17720
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org