jsancio commented on code in PR #12240:
URL: https://github.com/apache/kafka/pull/12240#discussion_r890595884
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -485,12 +541,36 @@ Iterator<UsableBroker> usableBrokers() {
id -> brokerRegistrations.get(id).rack());
}
+ /**
+ * Returns true if the broker is in fenced state; Returns false if it is
+ * not or if it does not exist.
+ */
public boolean unfenced(int brokerId) {
Review Comment:
Minor but I am pretty sure that we don't use this method anymore in
`src/main`. If so, I say that we just remove it.
##########
metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java:
##########
@@ -93,46 +93,35 @@ private BrokerRegistration getBrokerOrThrow(int brokerId,
long epoch, String act
public void replay(FenceBrokerRecord record) {
BrokerRegistration broker = getBrokerOrThrow(record.id(),
record.epoch(), "fence");
- changedBrokers.put(record.id(),
Optional.of(broker.cloneWithFencing(true)));
+ changedBrokers.put(record.id(), broker.maybeCloneWith(
+ BrokerRegistrationFencingChange.UNFENCE.asBoolean(),
+ Optional.empty()
+ ));
}
public void replay(UnfenceBrokerRecord record) {
BrokerRegistration broker = getBrokerOrThrow(record.id(),
record.epoch(), "unfence");
- changedBrokers.put(record.id(),
Optional.of(broker.cloneWithFencing(false)));
+ changedBrokers.put(record.id(), broker.maybeCloneWith(
+ BrokerRegistrationFencingChange.FENCE.asBoolean(),
+ Optional.empty()
+ ));
Review Comment:
We have an inflight change that changes these enum values around. We need to
be careful in the order we merge this PRs:
https://github.com/apache/kafka/pull/12236#issuecomment-1147912524
##########
metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java:
##########
@@ -213,12 +230,30 @@ public String toString() {
bld.append("}");
bld.append(", rack=").append(rack);
bld.append(", fenced=").append(fenced);
+ bld.append(", inControlledShutdown=").append(inControlledShutdown);
bld.append(")");
return bld.toString();
}
- public BrokerRegistration cloneWithFencing(boolean fencing) {
- return new BrokerRegistration(id, epoch, incarnationId, listeners,
- supportedFeatures, rack, fencing);
+ public Optional<BrokerRegistration> maybeCloneWith(
+ Optional<Boolean> fencingChange,
+ Optional<Boolean> inControlledShutdownChange
+ ) {
+ boolean newFenced = fencingChange.orElse(fenced);
+ boolean newInControlledShutdownChange =
inControlledShutdownChange.orElse(inControlledShutdown);
+
+ if (newFenced == fenced && newInControlledShutdownChange ==
inControlledShutdown)
+ return Optional.empty();
Review Comment:
Okay but we could just return `this`. Outside of `supportedFeatures`,
`BrokerRegistration` is immutable and it override `equals` and `hashCode` so
this should be safe.
I'll submit a minor PR to make `supportFeature` immutable.
##########
metadata/src/main/java/org/apache/kafka/image/MetadataImage.java:
##########
@@ -120,10 +121,16 @@ public AclsImage acls() {
}
public void write(Consumer<List<ApiMessageAndVersion>> out) {
+ // We use the minimum KRaft metadata version if this image does
+ // not have a specific version set.
+ MetadataVersion metadataVersion = features.metadataVersion();
+ if (metadataVersion.equals(MetadataVersion.UNINITIALIZED)) {
+ metadataVersion = MetadataVersion.IBP_3_0_IV1;
+ }
Review Comment:
I think we need a similar change for the `ClusterControlManager` iterator.
If I am not mistaken it is possible for the `FeatureControlManager` to return
`UNINITIALIZED` in both active controllers and inactive controllers.
Should we document this default version in `MetadataVersion` instead?
cc @mumrah
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
