root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-18844 Stale features information in QuorumController#registerBroker (#19058) 

In #16848, we added `kraft.version` to finalized features and got finalized features outside controller event handling thread. This may make finalized features stale when processing `registerBroker` event. Also, some cases like `QuorumControllerTest.testBalancePartitionLeaders` become flaky cause of outdated MV. This PR moves finalized features back to controller event handling thread to avoid the error.

Reviewers: Ismael Juma <ijuma@apache.org>, Jun Rao <junrao@gmail.com>, Colin P. McCabe <cmccabe@apache.org>, Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
PoAn Yang 2025-03-04 15:42:51 +08:00 committed by GitHub
parent 4d0216dd9f
commit 1ec1043d51
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
View File

@ -2251,11 +2251,12 @@ public final class QuorumController implements Controller {
ControllerRequestContext context,
BrokerRegistrationRequestData request
) {
// populate finalized features map with latest known kraft version for validation
Map<String, Short> controllerFeatures = new HashMap<>(featureControl.finalizedFeatures(Long.MAX_VALUE).featureMap());
controllerFeatures.put(KRaftVersion.FEATURE_NAME, raftClient.kraftVersion().featureLevel());
return appendWriteEvent("registerBroker", context.deadlineNs(),
() -> {
// Read and write data in the controller event handling thread to avoid stale information.
Map<String, Short> controllerFeatures = new HashMap<>(featureControl.finalizedFeatures(Long.MAX_VALUE).featureMap());
// Populate finalized features map with latest known kraft version for validation.
controllerFeatures.put(KRaftVersion.FEATURE_NAME, raftClient.kraftVersion().featureLevel());
ControllerResult<BrokerRegistrationReply> result = clusterControl.
registerBroker(request, offsetControl.nextWriteOffset(),
new FinalizedControllerFeatures(controllerFeatures, Long.MAX_VALUE));