mumrah commented on code in PR #13372:
URL: https://github.com/apache/kafka/pull/13372#discussion_r1135709978
##########
metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java:
##########
@@ -132,26 +132,45 @@ private boolean isControllerQuorumReadyForMigration() {
return true;
}
+ private boolean imageDoesNotContainAllBrokers(MetadataImage image,
Set<Integer> brokerIds) {
+ for (BrokerRegistration broker : image.cluster().brokers().values()) {
+ if (broker.isMigratingZkBroker()) {
+ brokerIds.remove(broker.id());
+ }
+ }
+ return !brokerIds.isEmpty();
+ }
+
private boolean areZkBrokersReadyForMigration() {
if (image == MetadataImage.EMPTY) {
// TODO maybe add WAIT_FOR_INITIAL_METADATA_PUBLISH state to avoid
this kind of check?
log.info("Waiting for initial metadata publish before checking if
Zk brokers are registered.");
return false;
}
- Set<Integer> zkRegisteredZkBrokers =
zkMigrationClient.readBrokerIdsFromTopicAssignments();
- for (BrokerRegistration broker : image.cluster().brokers().values()) {
- if (broker.isMigratingZkBroker()) {
- zkRegisteredZkBrokers.remove(broker.id());
- }
+
+ // First check the brokers registered in ZK
+ Set<Integer> zkBrokerRegistrations = new
HashSet<>(zkMigrationClient.readBrokerIds());
Review Comment:
Ah, yea this is needed because the Set returned by the client was an
immutable Scala thing originally. I'll change the client to return a mutable set
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
