You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "C0urante (via GitHub)" <gi...@apache.org> on 2024/07/12 14:46:10 UTC
[PR] KAFKA-17105: Prevent redundant restarts for newly-created connectors [kafka]
C0urante opened a new pull request, #16581:
URL: https://github.com/apache/kafka/pull/16581
[Jira](https://issues.apache.org/jira/browse/KAFKA-17105)
Will update description if testing indicates this fixes the flaky test cases in `OffsetsApiIntegrationTest` related to zombie sink tasks.
### 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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #16581:
URL: https://github.com/apache/kafka/pull/16581#issuecomment-2233992027
Thanks Greg!
CI looks good on all three nodes that have completed (three flaky test failures in unrelated modules), and the build (but not test) has completed on the only other node. Merging...
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante merged PR #16581:
URL: https://github.com/apache/kafka/pull/16581
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #16581:
URL: https://github.com/apache/kafka/pull/16581#issuecomment-2229540656
@gharris1727 here's another flaky test fix if you have the bandwidth. It even got a green build!
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on code in PR #16581:
URL: https://github.com/apache/kafka/pull/16581#discussion_r1680185345
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1579,6 +1581,55 @@ public void testConnectorConfigAdded() {
verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
}
+ @ParameterizedTest
+ @ValueSource(shorts = {CONNECT_PROTOCOL_V0, CONNECT_PROTOCOL_V1, CONNECT_PROTOCOL_V2})
+ public void testConnectorConfigDetectedAfterLeaderAlreadyAssigned(short protocolVersion) {
+ connectProtocolVersion = protocolVersion;
+
+ // If a connector was added, we need to rebalance
+ when(worker.isSinkConnector(CONN1)).thenReturn(Boolean.TRUE);
+ when(member.memberId()).thenReturn("member");
+ when(member.currentProtocolVersion()).thenReturn(protocolVersion);
+
+ // join, no configs so no need to catch up on config topic
+ expectRebalance(-1, Collections.emptyList(), Collections.emptyList());
+ expectMemberPoll();
+
+ herder.tick(); // join
+
+ // Checks for config updates and starts rebalance
+ configUpdateListener.onConnectorConfigUpdate(CONN1); // read updated config
+ expectConfigRefreshAndSnapshot(SNAPSHOT);
+
+ // Rebalance will be triggered when the new config is detected
+ // TODO: This is still buggy; we're requesting a rebalance for a connector
+ // that's already been assigned to us, which is clearly unnecessary.
+ // We might consider tracking the offsets of records we've read from the
+ // config topic and only triggering rebalances if we see something that's
+ // greater than the offset included by the leader in the latest assignment
Review Comment:
Should this be a follow-up ticket? If there is already one, can you mention it in the comment?
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #16581:
URL: https://github.com/apache/kafka/pull/16581#issuecomment-2233739849
Thanks for the review Greg! Regarding your questions:
> [Do you think that this] change to the ConfigUpdateListener makes the logic in updateConfigsWithEager unnecessary?
I believe it's still necessary, though only for a pretty rare case. It's possible for a new connector config to be written and an existing connector config to be updated at the same time. This would require a split-brain scenario where two workers believe they are the leader, but without the use of a transactional producer by the leader it isn't impossible.
In that case, we would still want to clear the `connectorConfigUpdates` field before participating in the rebalance, since all connectors and tasks assigned to the worker would stopped be in preparation for the rebalance.
> [Do you think that ] we should have clearing logic in updateConfigsWithIncrementalCooperative?
I think the lack of clearing logic is intentional. If we consider the same scenario as above, we'd want to hold onto `connectorConfigUpdates` since we wouldn't be stopping all connectors assigned to the worker in preparation for the rebalance.
> [Do you think that ] we could/should perform this cleanup some other time?
Given that the cases covered above seem reasonably addressed by the current patch, and that all unit and integration tests indicate a lack of regression, I'm hesitant to pursue a larger refactoring. Were there any other alternatives you had in mind?
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #16581:
URL: https://github.com/apache/kafka/pull/16581#issuecomment-2226344625
After validating this fix and https://github.com/apache/kafka/pull/16567 in https://github.com/apache/kafka/pull/16582 (using https://github.com/apache/kafka/pull/16583 as a control group), it seems like the two fixes should go a long way towards eliminating the flakiness we've seen in the `OffsetsApiIntegrationTest`.
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #16581:
URL: https://github.com/apache/kafka/pull/16581#discussion_r1681353545
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java:
##########
@@ -1579,6 +1581,55 @@ public void testConnectorConfigAdded() {
verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
}
+ @ParameterizedTest
+ @ValueSource(shorts = {CONNECT_PROTOCOL_V0, CONNECT_PROTOCOL_V1, CONNECT_PROTOCOL_V2})
+ public void testConnectorConfigDetectedAfterLeaderAlreadyAssigned(short protocolVersion) {
+ connectProtocolVersion = protocolVersion;
+
+ // If a connector was added, we need to rebalance
+ when(worker.isSinkConnector(CONN1)).thenReturn(Boolean.TRUE);
+ when(member.memberId()).thenReturn("member");
+ when(member.currentProtocolVersion()).thenReturn(protocolVersion);
+
+ // join, no configs so no need to catch up on config topic
+ expectRebalance(-1, Collections.emptyList(), Collections.emptyList());
+ expectMemberPoll();
+
+ herder.tick(); // join
+
+ // Checks for config updates and starts rebalance
+ configUpdateListener.onConnectorConfigUpdate(CONN1); // read updated config
+ expectConfigRefreshAndSnapshot(SNAPSHOT);
+
+ // Rebalance will be triggered when the new config is detected
+ // TODO: This is still buggy; we're requesting a rebalance for a connector
+ // that's already been assigned to us, which is clearly unnecessary.
+ // We might consider tracking the offsets of records we've read from the
+ // config topic and only triggering rebalances if we see something that's
+ // greater than the offset included by the leader in the latest assignment
Review Comment:
Sure, I've filed https://issues.apache.org/jira/browse/KAFKA-17155 to track.
--
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-17105: Prevent redundant restarts for newly-created connectors [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #16581:
URL: https://github.com/apache/kafka/pull/16581#issuecomment-2233834388
Thanks for explaining the situation which makes the updateConfigs* functions accurate. I agree that the current PR is sufficient to fix this bug.
> Were there any other alternatives you had in mind?
I think the ad-hoc state management of the DistributedHerder is partially to blame for hiding this bug, and also made it harder to review this fix. I haven't fully formed a vision of what the refactor would look like, but I opened a ticket to look into it later: https://issues.apache.org/jira/browse/KAFKA-17157
--
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