ahuang98 commented on code in PR #19454:
URL: https://github.com/apache/kafka/pull/19454#discussion_r2042995867
##########
metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java:
##########
@@ -980,131 +979,6 @@ 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)));
-
- 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().
Review Comment:
honestly, we could keep this one as well if we just modified this line to
`assertThrows(DuplicateBrokerRegistrationException.class, () -> ...)`
--
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
