KAFKA-19130: Do not add fenced brokers to BrokerRegistrationTracker on startup (apache#19454)

When the controller starts up (or becomes active after being inactive), we add all of the registered brokers to BrokerRegistrationTracker so that they will not be accidentally fenced the next time we are looking for a broker to fence. We do this because the state in BrokerRegistrationTracker is "soft state" (it doesn't appear in the metadata log), and the newly active controller starts off with no soft state. (Its soft state will be populated by the brokers sending heartbeat requests to it over time.)

In the case of fenced brokers, we are not worried about accidentally fencing the broker due to it being missing from
BrokerRegistrationTracker for a while (it's already fenced). Therefore, it should be reasonable to just not add fenced brokers to the tracker initially.

One case where this change will have a positive impact is for people running single-node demonstration clusters in combined KRaft mode. In that case, when the single-node cluster is taken down and restarted, it currently will have to wait about 9 seconds for the broker to come up and re-register. With this change, the broker should be able to re-register immediately (assuming the previous shutdown happened cleanly through controller shutdown.)

One possible negative impact is that if there is a controller failover, it will open a small window where a broker with the same ID as a fenced broker could re-register. However, our detection of duplicate broker IDs is best-effort (and duplicate broker IDs are an administrative mistake), so this downside seems acceptable.

Reviewers: Alyssa Huang <[email protected]>, José Armando García Sancio <[email protected]>

Author: cmccabe

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -309,8 +309,10 @@ public void activate() {
         long nowNs = time.nanoseconds();
         for (BrokerRegistration registration : brokerRegistrations.values()) {
             heartbeatManager.register(registration.id(), registration.fenced());
-            heartbeatManager.tracker().updateContactTime(
-                new BrokerIdAndEpoch(registration.id(), registration.epoch()), nowNs);
+            if (!registration.fenced()) {
+                heartbeatManager.tracker().updateContactTime(
+                    new BrokerIdAndEpoch(registration.id(), registration.epoch()), nowNs);
+            }
         }
     }
 
@@ -353,9 +355,9 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
         if (existing != null) {
             prevIncarnationId = existing.incarnationId();
             storedBrokerEpoch = existing.epoch();
-            if (heartbeatManager.hasValidSession(brokerId, existing.epoch()) && !existing.inControlledShutdown()) {
+            if (heartbeatManager.hasValidSession(brokerId, existing.epoch())) {
                 if (!request.incarnationId().equals(prevIncarnationId)) {
-                    throw new DuplicateBrokerRegistrationException("Another broker is registered with that broker id. If the broker " + 
+                    throw new DuplicateBrokerRegistrationException("Another broker is registered with that broker id. If the broker " +
                             "was recently restarted this should self-resolve once the heartbeat manager expires the broker's session.");
                 }
             }

metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java

@@ -968,9 +968,10 @@ public void testBrokerContactTimesAreUpdatedOnClusterControlActivation() {
         clusterControl.replay(new RegisterBrokerRecord().
             setBrokerEpoch(123).
             setBrokerId(1).
+            setFenced(false).
             setLogDirs(List.of(Uuid.fromString("TyNK6XSSQJaJc2q9uflNHg"))), 10005);
         clusterControl.activate();
-        assertEquals(OptionalLong.of(1000L), clusterControl.heartbeatManager().tracker().
+        assertEquals(OptionalLong.empty(), clusterControl.heartbeatManager().tracker().
             contactTime(new BrokerIdAndEpoch(0, 100)));
         assertEquals(OptionalLong.of(1000L), clusterControl.heartbeatManager().tracker().
             contactTime(new BrokerIdAndEpoch(1, 123)));
@@ -1068,41 +1069,40 @@ public void testDuplicateBrokerRegistrationWithInactiveBroker() {
         clusterControl.activate();
         clusterControl.heartbeatManager().maybeUpdateControlledShutdownOffset(1, 20002);
 
-        assertEquals(OptionalLong.of(1000L), clusterControl.heartbeatManager().tracker().
+        assertEquals(OptionalLong.empty(), clusterControl.heartbeatManager().tracker().
             contactTime(new BrokerIdAndEpoch(0, 100)));
         assertEquals(OptionalLong.of(1000L), clusterControl.heartbeatManager().tracker().
             contactTime(new BrokerIdAndEpoch(1, 200)));
 
         time.sleep(brokerSessionTimeoutMs / 2);
-        assertThrows(DuplicateBrokerRegistrationException.class, () ->
-            clusterControl.registerBroker(new BrokerRegistrationRequestData().
-                    setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
-                    setBrokerId(0).
-                    setLogDirs(List.of(Uuid.fromString("yJGxmjfbQZSVFAlNM3uXZg"))).
-                    setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
-                        Set.of(new BrokerRegistrationRequestData.Feature().
-                            setName(MetadataVersion.FEATURE_NAME).
-                            setMinSupportedVersion(MetadataVersion.MINIMUM_VERSION.featureLevel()).
-                            setMaxSupportedVersion(MetadataVersion.LATEST_PRODUCTION.featureLevel())).iterator())).
-                    setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")),
-                101L,
-                finalizedFeatures,
-                false));
-        // new registration should succeed immediatelly only if the broker is in controlled shutdown, 
-        // even if the last heartbeat was within the session timeout 
         clusterControl.registerBroker(new BrokerRegistrationRequestData().
                 setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
-                setBrokerId(1).
-                setLogDirs(List.of(Uuid.fromString("b66ybsWIQoygs01vdjH07A"))).
+                setBrokerId(0).
+                setLogDirs(List.of(Uuid.fromString("yJGxmjfbQZSVFAlNM3uXZg"))).
                 setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
                     Set.of(new BrokerRegistrationRequestData.Feature().
                         setName(MetadataVersion.FEATURE_NAME).
                         setMinSupportedVersion(MetadataVersion.MINIMUM_VERSION.featureLevel()).
                         setMaxSupportedVersion(MetadataVersion.LATEST_PRODUCTION.featureLevel())).iterator())).
-                setIncarnationId(Uuid.fromString("vZKYST0pSA2HO5x_6hoO2Q")),
-            201L,
+                setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")),
+            101L,
             finalizedFeatures,
             false);
+        assertThrows(DuplicateBrokerRegistrationException.class, () -> {
+            clusterControl.registerBroker(new BrokerRegistrationRequestData().
+                    setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
+                    setBrokerId(1).
+                    setLogDirs(List.of(Uuid.fromString("b66ybsWIQoygs01vdjH07A"))).
+                    setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
+                        Set.of(new BrokerRegistrationRequestData.Feature().
+                            setName(MetadataVersion.FEATURE_NAME).
+                            setMinSupportedVersion(MetadataVersion.MINIMUM_VERSION.featureLevel()).
+                            setMaxSupportedVersion(MetadataVersion.LATEST_PRODUCTION.featureLevel())).iterator())).
+                    setIncarnationId(Uuid.fromString("vZKYST0pSA2HO5x_6hoO2Q")),
+                201L,
+                finalizedFeatures,
+                false);
+        });
     }
 
     private FeatureControlManager createFeatureControlManager() {