splett2 commented on code in PR #19296:
URL: https://github.com/apache/kafka/pull/19296#discussion_r2017080238
##########
metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java:
##########
@@ -978,6 +980,129 @@ public void
testBrokerContactTimesAreUpdatedOnClusterControlActivation() {
contactTime(new BrokerIdAndEpoch(2, 100)));
}
+ @Test
+ public void testDuplicateBrokerRegistrationWithActiveOldBroker() {
+ // active here means brokerHeartbeatManager last recorded the broker
as unfenced and not in controlled shutdown
+ long brokerSessionTimeoutMs = 1000;
+ MockTime time = new MockTime(0L, 20L, 1000L);
+ FinalizedControllerFeatures finalizedFeatures = new
FinalizedControllerFeatures(
+ Map.of(MetadataVersion.FEATURE_NAME,
MetadataVersion.LATEST_PRODUCTION.featureLevel()), 456L);
+ ClusterControlManager clusterControl = new
ClusterControlManager.Builder().
+ setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
+ setFeatureControlManager(createFeatureControlManager()).
+ setBrokerShutdownHandler((brokerId, isCleanShutdown, records) -> {
}).
+
setSessionTimeoutNs(TimeUnit.MILLISECONDS.toNanos(brokerSessionTimeoutMs)).
+ setTime(time).
+ build();
+ clusterControl.replay(new RegisterBrokerRecord().
+ setBrokerEpoch(100).
+ setBrokerId(0).
+ setLogDirs(List.of(Uuid.fromString("Mj3CW3OSRi29cFeNJlXuAQ"))).
+ setFenced(false), 10002);
+ clusterControl.activate();
+ assertEquals(OptionalLong.of(1000L),
clusterControl.heartbeatManager().tracker().
+ contactTime(new BrokerIdAndEpoch(0, 100)));
+
+ // while session is still valid for old broker, duplicate requests
should fail
+ time.sleep(brokerSessionTimeoutMs / 2);
+ assertThrows(DuplicateBrokerRegistrationException.class, () ->
+ clusterControl.registerBroker(new BrokerRegistrationRequestData().
+ setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
+ setBrokerId(0).
+
setLogDirs(List.of(Uuid.fromString("TyNK6XSSQJaJc2q9uflNHg"))).
+ 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));
+
+ // if session expires for broker, even if the broker was active the
new registration will succeed
+ time.sleep(brokerSessionTimeoutMs);
+ clusterControl.registerBroker(new BrokerRegistrationRequestData().
+ setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
+ setBrokerId(0).
+ setLogDirs(List.of(Uuid.fromString("TyNK6XSSQJaJc2q9uflNHg"))).
+ 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);
+ }
+
+ @Test
+ public void testDuplicateBrokerRegistrationWithInactiveBroker() {
+ // inactive here means brokerHeartbeatManager last recorded the broker
as fenced or in controlled shutdown
+ long brokerSessionTimeoutMs = 1000;
+ MockTime time = new MockTime(0L, 20L, 1000L);
+ FinalizedControllerFeatures finalizedFeatures = new
FinalizedControllerFeatures(
+ Map.of(MetadataVersion.FEATURE_NAME,
MetadataVersion.LATEST_PRODUCTION.featureLevel()), 456L);
+ ClusterControlManager clusterControl = new
ClusterControlManager.Builder().
+ setClusterId("pjvUwj3ZTEeSVQmUiH3IJw").
+ setFeatureControlManager(createFeatureControlManager()).
+ setBrokerShutdownHandler((brokerId, isCleanShutdown, records) -> {
}).
+
setSessionTimeoutNs(TimeUnit.MILLISECONDS.toNanos(brokerSessionTimeoutMs)).
+ setTime(time).
+ build();
+ // first broker is fenced
+ clusterControl.replay(new RegisterBrokerRecord().
+ setBrokerEpoch(100).
+ setBrokerId(0).
+ setLogDirs(List.of(Uuid.fromString("Mj3CW3OSRi29cFeNJlXuAQ"))).
+ setFenced(true).
+ setInControlledShutdown(false), 10002);
+ // second broker is in controlled shutdown
+ clusterControl.replay(new RegisterBrokerRecord().
+ setBrokerEpoch(200).
+ setBrokerId(1).
+ setLogDirs(List.of(Uuid.fromString("TyNK6XSSQJaJc2q9uflNHg"))).
+ setFenced(false).
+ setInControlledShutdown(true), 20002);
+ clusterControl.activate();
+
clusterControl.heartbeatManager().maybeUpdateControlledShutdownOffset(1, 20002);
+
+ assertEquals(OptionalLong.of(1000L),
clusterControl.heartbeatManager().tracker().
+ contactTime(new BrokerIdAndEpoch(0, 100)));
+ assertEquals(OptionalLong.of(1000L),
clusterControl.heartbeatManager().tracker().
+ contactTime(new BrokerIdAndEpoch(1, 200)));
+
+ // even if session is still valid for old brokers, new registrations
will succeed
Review Comment:
nit: I think the wording here should be something along the lines of:
> new registrations should succeed if a broker is fenced or in controlled
shutdown, even if the last heartbeat was within the session timeout
--
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
