You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/08/09 15:53:53 UTC
[GitHub] [kafka] niket-goel opened a new pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
niket-goel opened a new pull request #11191:
URL: https://github.com/apache/kafka/pull/11191
This is necessary as the effect of each fencing is visible to the
controller one the corresponding record for it has been generated and
applied. Processing stale brokers one at a time ensures that subsequent
fencing records are generated correctly with the correct view of the system.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r685530522
##########
File path: metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
##########
@@ -858,7 +858,7 @@ private void rescheduleMaybeFenceStaleBrokers() {
return;
}
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs, () -> {
- ControllerResult<Void> result = replicationControl.maybeFenceStaleBrokers();
+ ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
rescheduleMaybeFenceStaleBrokers();
Review comment:
I agree that a test covering this would be good. Will add that to the next revision.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687262045
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
Review comment:
The topic creation actually fails here. We try the create again after unfencing the brokers.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
Review comment:
Makes sense, will modify.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r686498956
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +175,127 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeout = 1L;
+ Long sleepMillis = (sessionTimeout * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeout))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ boolean fencingComplete;
+ Long waitIterations = 0L;
+ do {
+ fencingComplete = true;
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ fencingComplete = false;
+ }
+ }
+ Thread.sleep(1000);
+ waitIterations++;
+
+ if (waitIterations >= sessionTimeout * 3) {
+ assertTrue(false, "Fencing of brokers did not process within expected time");
+ }
+ } while (!fencingComplete);
Review comment:
Will check this out. Thanks for the pointer.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +175,127 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeout = 1L;
+ Long sleepMillis = (sessionTimeout * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeout))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ boolean fencingComplete;
+ Long waitIterations = 0L;
+ do {
+ fencingComplete = true;
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ fencingComplete = false;
+ }
+ }
+ Thread.sleep(1000);
+ waitIterations++;
+
+ if (waitIterations >= sessionTimeout * 3) {
+ assertTrue(false, "Fencing of brokers did not process within expected time");
Review comment:
Thanks (new to the project and JAVA :) )
##########
File path: metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -208,6 +212,24 @@ void unfenceBrokers(Integer... brokerIds) throws Exception {
}
}
+ void fenceBrokers(Set<Integer> brokerIds) throws Exception {
+ time.sleep(BROKER_SESSION_TIMEOUT_MS);
Review comment:
Good call.. Although I would not do it in the common utility function.
Will do it in the test method where this should hold.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687186647
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
Review comment:
This part is a little mysterious to me. We return an error, but the topic is created anyway? That seems surprising.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
Review comment:
nit: it's a small thing, but the assertion failure message is more useful if we use the `Errors` type.
```java
assertEquals(Errors.NONE, Errors.forCode(createTopicsResponseData.topics().find("foo").errorCode()));
```
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
Review comment:
nit: maybe we can use milliseconds consistently and leave out some unit conversions below.
Also, could this and `sleepMillis` be a `long` instead of `Long`?
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
Review comment:
Perhaps we can use `Uuid.randomUuid`? It's a little weird for all brokers to have the same incarnationId.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
Review comment:
Is there a point to setting `min.insync.replicas` in this test? I am wondering why we don't just reuse the original create request.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
Review comment:
Maybe a little subjective, but I think the test case would be more readable if we list these brokers directly. For example:
```java
List<Integer> allBrokers = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> brokersToKeepUnfenced = Arrays.asList(1);
```
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
Review comment:
nit: we can combine these two `try` blocks:
```java
try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty());
QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
...
```
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
##########
@@ -36,16 +41,29 @@
private final List<QuorumController> controllers;
private final LocalLogManagerTestEnv logEnv;
- public QuorumControllerTestEnv(LocalLogManagerTestEnv logEnv,
- Consumer<QuorumController.Builder> builderConsumer)
- throws Exception {
+ public QuorumControllerTestEnv(
+ LocalLogManagerTestEnv logEnv,
+ Consumer<QuorumController.Builder> builderConsumer
+ ) throws Exception {
+ this(logEnv, builderConsumer, Optional.empty());
+ }
+
+ public QuorumControllerTestEnv(
+ LocalLogManagerTestEnv logEnv,
+ Consumer<Builder> builderConsumer,
+ Optional<Long> sessionTimeoutSeconds
Review comment:
nit: as above, can we use milliseconds? It's the most conventional unit we use in the code for configurations representing durations.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ TestUtils.waitForCondition(() -> {
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ return false;
+ }
+ }
+ return true;
+ }, sessionTimeoutSec * 2 * 1000,
+ "Fencing of brokers did not process within expected time"
+ );
+
+ // At this point the brokers we want fenced are fenced.
+ // Send another heartbeat to the brokers we want to keep alive
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ int[] expectedIsr = new int[brokersToKeepUnfenced];
Review comment:
If we create the topic with an explicit assignment, then I think we can remove this logic to build the expected ISR since we would be able to assert the
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r686463360
##########
File path: metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -208,6 +212,24 @@ void unfenceBrokers(Integer... brokerIds) throws Exception {
}
}
+ void fenceBrokers(Set<Integer> brokerIds) throws Exception {
+ time.sleep(BROKER_SESSION_TIMEOUT_MS);
Review comment:
nit: we can assert the `fencedBrokerIds` is empty before sleep to timeout session.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -1029,6 +1051,24 @@ public void testCreatePartitionsFailsWithManualAssignmentWithAllFenced() throws
ctx.replicationControl.getPartition(fooId, 1));
}
+ @Test
+ public void testFenceMultipleBrokers() throws Exception {
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext();
+ ReplicationControlManager replication = ctx.replicationControl;
+ ctx.registerBrokers(0, 1, 2, 3, 4);
+ ctx.unfenceBrokers(0, 1, 2, 3, 4);
+
+ Uuid fooId = ctx.createTestTopic("foo", new int[][]{
+ new int[]{1, 2, 3}, new int[]{2, 3, 4}, new int[]{0, 2, 1}}).topicId();
+
+ ctx.fenceBrokers(Utils.mkSet(2, 3));
+
+ PartitionRegistration partition0 = replication.getPartition(fooId, 0);
+ assertArrayEquals(new int[]{1, 2, 3}, partition0.replicas);
+ assertArrayEquals(new int[]{1}, partition0.isr);
+ assertEquals(1, partition0.leader);
Review comment:
I think we can also verify other partitions, not just partition 0. ex:
```java
PartitionRegistration partition0 = replication.getPartition(fooId, 0);
PartitionRegistration partition1 = replication.getPartition(fooId, 1);
PartitionRegistration partition2 = replication.getPartition(fooId, 2);
assertArrayEquals(new int[]{1, 2, 3}, partition0.replicas);
assertArrayEquals(new int[]{1}, partition0.isr);
assertEquals(1, partition0.leader);
assertArrayEquals(new int[]{2, 3, 4}, partition1.replicas);
assertArrayEquals(new int[]{4}, partition1.isr);
assertEquals(4, partition1.leader);
assertArrayEquals(new int[]{0, 2, 1}, partition2.replicas);
assertArrayEquals(new int[]{0, 1}, partition2.isr);
assertNotEquals(2, partition2.leader);
```
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +175,127 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeout = 1L;
+ Long sleepMillis = (sessionTimeout * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeout))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ boolean fencingComplete;
+ Long waitIterations = 0L;
+ do {
+ fencingComplete = true;
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ fencingComplete = false;
+ }
+ }
+ Thread.sleep(1000);
+ waitIterations++;
+
+ if (waitIterations >= sessionTimeout * 3) {
+ assertTrue(false, "Fencing of brokers did not process within expected time");
Review comment:
You can call `fail` directly: `fail("Fencing of brokers did not process within expected time");`
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +175,127 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeout = 1L;
+ Long sleepMillis = (sessionTimeout * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeout))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ boolean fencingComplete;
+ Long waitIterations = 0L;
+ do {
+ fencingComplete = true;
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ fencingComplete = false;
+ }
+ }
+ Thread.sleep(1000);
+ waitIterations++;
+
+ if (waitIterations >= sessionTimeout * 3) {
+ assertTrue(false, "Fencing of brokers did not process within expected time");
+ }
+ } while (!fencingComplete);
Review comment:
Maybe you can replace this `while` loop with a util function `TestUtils.waitForCondition`. Ex:
`TestUtils.waitForCondition(() -> fenceBrokers(), 3000, "Fencing of brokers did not process within expected time");`
Just that it only accepted the `timeout` value, not the `waitIterations` like you did. FYI
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687047710
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
##########
@@ -161,6 +163,14 @@ public void deactivate() {
return brokerRegistrations;
}
+ Set<Integer> fencedBrokerIds() {
Review comment:
Looks like this method is only used for tests. How about moving this to `ReplicationControlManagerTest`?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java
##########
@@ -420,22 +419,22 @@ long nextCheckTimeNs() {
}
/**
- * Find the stale brokers which haven't heartbeated in a long time, and which need to
- * be fenced.
+ * Check if the oldest broker to have hearbeated has already violated the
+ * sessionTimeoutNs timeout and needs to be fenced.
*
- * @return A list of node IDs.
+ * @return An Optional broker node id.
*/
- List<Integer> findStaleBrokers() {
- List<Integer> nodes = new ArrayList<>();
+ Optional<Integer> findOneStaleBroker() {
BrokerHeartbeatStateIterator iterator = unfenced.iterator();
- while (iterator.hasNext()) {
+ if (iterator.hasNext()) {
Review comment:
Isn't `unfenced.first()` suppose to return the oldest heartbeat? If so, can use that method and check for `null`?
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
Review comment:
Can we remove this sleep? What are we trying to test with this sleep?
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +174,89 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ List<Integer> allBrokers = Arrays.asList(1, 2, 3, 4, 5);
+ List<Integer> brokersToKeepUnfenced = Arrays.asList(1);
+ List<Integer> brokersToFence = Arrays.asList(2, 3, 4, 5);
+ short replicationFactor = 5;
+ long sessionTimeoutMillis = 1000;
+
+ try (
+ LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty());
+ QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv(
+ logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutMillis));
+ ) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+
+ for (Integer brokerId : allBrokers) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.randomUuid()).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered and should still be fenced
+ allBrokers.forEach(brokerId -> {
+ assertFalse(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been fenced");
+ });
+
+ // Unfence all brokers and create a topic foo
+ sendBrokerheartbeat(active, allBrokers, brokerEpochs);
+ CreateTopicsRequestData createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE, Errors.forCode(createTopicsResponseData.topics().find("foo").errorCode()));
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+
+ // Fence some of the brokers
+ TestUtils.waitForCondition(() -> {
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (Integer brokerId : brokersToFence) {
+ if (active.replicationControl().isBrokerUnfenced(brokerId)) {
+ return false;
+ }
+ }
+ return true;
+ }, sessionTimeoutMillis * 3,
+ "Fencing of brokers did not process within expected time"
+ );
+
+ // Send another heartbeat to the brokers we want to keep alive
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+
+ // At this point only the brokers we want fenced should be fenced.
+ brokersToKeepUnfenced.forEach(brokerId -> {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ });
+ brokersToFence.forEach(brokerId -> {
+ assertFalse(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been fenced");
+ });
+
+ // Verify the isr and leaders for the topic partition
+ int[] expectedIsr = {1};
+ int[] isrFoo = active.replicationControl().getPartition(topicIdFoo, 0).isr;
+
+ assertTrue(Arrays.equals(isrFoo, expectedIsr),
+ "The ISR for topic foo was " + Arrays.toString(isrFoo) +
+ ". It is expected to be " + Arrays.toString(expectedIsr));
Review comment:
If you use `assertEquals` it will compare the content of the arrays and print their value if they don't match.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -1029,6 +1052,36 @@ public void testCreatePartitionsFailsWithManualAssignmentWithAllFenced() throws
ctx.replicationControl.getPartition(fooId, 1));
}
+ @Test
+ public void testFenceMultipleBrokers() throws Exception {
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext();
+ ReplicationControlManager replication = ctx.replicationControl;
+ ctx.registerBrokers(0, 1, 2, 3, 4);
+ ctx.unfenceBrokers(0, 1, 2, 3, 4);
+
+ Uuid fooId = ctx.createTestTopic("foo", new int[][]{
+ new int[]{1, 2, 3}, new int[]{2, 3, 4}, new int[]{0, 2, 1}}).topicId();
+
+ assertTrue(ctx.clusterControl.fencedBrokerIds().isEmpty());
Review comment:
You can `assertEquals` against the empty `Set` (`Collections.emptySet()`). When it fails it will print the content of fenced broker ids.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ TestUtils.waitForCondition(() -> {
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ return false;
+ }
+ }
+ return true;
+ }, sessionTimeoutSec * 2 * 1000,
+ "Fencing of brokers did not process within expected time"
+ );
+
+ // At this point the brokers we want fenced are fenced.
+ // Send another heartbeat to the brokers we want to keep alive
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ int[] expectedIsr = new int[brokersToKeepUnfenced];
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ if (brokerId < brokersToKeepUnfenced) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ expectedIsr[brokerId] = brokerId;
+ } else {
+ assertFalse(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been fenced");
+ }
+ }
+
+ // Verify the isr and leaders for the topic partitions
+ int[] sortedIsrFoo = active.replicationControl().getPartition(topicIdFoo, 0).isr.clone();
+ int[] sortedIsrBar = active.replicationControl().getPartition(topicIdBar, 0).isr.clone();
+ Arrays.sort(sortedIsrFoo);
+ Arrays.sort(sortedIsrBar);
+ assertTrue(Arrays.equals(sortedIsrFoo, expectedIsr)
+ && Arrays.equals(sortedIsrBar, expectedIsr),
+ "The ISR for topic foo was " + Arrays.toString(sortedIsrFoo) +
+ " and for topic bar was " + Arrays.toString(sortedIsrBar) +
+ ". Both are expected to be " + Arrays.toString(expectedIsr));
Review comment:
You can split this into two calls to `assertEquals`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#issuecomment-897785278
@niket-goel I'm inclined to merge this as is so that we don't need to wait for another build. Perhaps you can submit a small follow-up to address @jsancio 's comments?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r685421603
##########
File path: metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
##########
@@ -858,7 +858,7 @@ private void rescheduleMaybeFenceStaleBrokers() {
return;
}
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs, () -> {
- ControllerResult<Void> result = replicationControl.maybeFenceStaleBrokers();
+ ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
rescheduleMaybeFenceStaleBrokers();
Review comment:
One thing I was considering is whether we should schedule the next check right away. For example, maybe we could check whether `result` contains a non-empty record set and reschedule immediately (or perhaps after a short backoff) if it does.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -925,14 +926,21 @@ ApiError electLeader(String topic, int partitionId, boolean uncleanOk,
return ControllerResult.of(records, null);
}
- ControllerResult<Void> maybeFenceStaleBrokers() {
+ ControllerResult<Void> maybeFenceOneStaleBroker() {
List<ApiMessageAndVersion> records = new ArrayList<>();
BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
List<Integer> staleBrokers = heartbeatManager.findStaleBrokers();
- for (int brokerId : staleBrokers) {
+ if (!staleBrokers.isEmpty()) {
+ // Even though multiple brokers can go stale at a time, we will process
+ // fencing one at a time so that the effect of fencing each broker is visible
+ // to the controller (memory) prior to processing the fencing of the next one
+ int brokerId = staleBrokers.get(0);
+ log.debug("Found broker(s) {} to be stale. Processing them one at a time",
+ Arrays.toString(staleBrokers.toArray()));
log.info("Fencing broker {} because its session has timed out.", brokerId);
handleBrokerFenced(brokerId, records);
heartbeatManager.fence(brokerId);
+
Review comment:
nit: remove newline
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -925,14 +926,21 @@ ApiError electLeader(String topic, int partitionId, boolean uncleanOk,
return ControllerResult.of(records, null);
}
- ControllerResult<Void> maybeFenceStaleBrokers() {
+ ControllerResult<Void> maybeFenceOneStaleBroker() {
List<ApiMessageAndVersion> records = new ArrayList<>();
BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
List<Integer> staleBrokers = heartbeatManager.findStaleBrokers();
- for (int brokerId : staleBrokers) {
+ if (!staleBrokers.isEmpty()) {
Review comment:
If we are only expiring one, maybe we can change `findStaleBrokers` to `findStaleBroker` and return `Optional<Integer>` instead of `List<Integer>`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687194150
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ TestUtils.waitForCondition(() -> {
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ return false;
+ }
+ }
+ return true;
+ }, sessionTimeoutSec * 2 * 1000,
+ "Fencing of brokers did not process within expected time"
+ );
+
+ // At this point the brokers we want fenced are fenced.
+ // Send another heartbeat to the brokers we want to keep alive
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ int[] expectedIsr = new int[brokersToKeepUnfenced];
Review comment:
If we create the topic with an explicit assignment, then I think we can remove this logic to build the expected ISR since we would be able to assert the expected ISR directly.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#issuecomment-896508951
@showuon -- Thanks for looking at the PR. Will address the comments and publish an update.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r685463058
##########
File path: metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
##########
@@ -858,7 +858,7 @@ private void rescheduleMaybeFenceStaleBrokers() {
return;
}
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs, () -> {
- ControllerResult<Void> result = replicationControl.maybeFenceStaleBrokers();
+ ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
rescheduleMaybeFenceStaleBrokers();
Review comment:
It sounds like you are saying there may still be a race condition if the next expiration is scheduled too quickly. However, I think the way the code works, we are guaranteed to have this result reflected before the next scheduled operation can trigger. It would be good to have a test covering this scenario.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r685438382
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -925,14 +926,21 @@ ApiError electLeader(String topic, int partitionId, boolean uncleanOk,
return ControllerResult.of(records, null);
}
- ControllerResult<Void> maybeFenceStaleBrokers() {
+ ControllerResult<Void> maybeFenceOneStaleBroker() {
List<ApiMessageAndVersion> records = new ArrayList<>();
BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
List<Integer> staleBrokers = heartbeatManager.findStaleBrokers();
- for (int brokerId : staleBrokers) {
+ if (!staleBrokers.isEmpty()) {
Review comment:
That is fair. Let me do that (wasn't sure if we wanted to add another method for this case).
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -925,14 +926,21 @@ ApiError electLeader(String topic, int partitionId, boolean uncleanOk,
return ControllerResult.of(records, null);
}
- ControllerResult<Void> maybeFenceStaleBrokers() {
+ ControllerResult<Void> maybeFenceOneStaleBroker() {
List<ApiMessageAndVersion> records = new ArrayList<>();
BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
List<Integer> staleBrokers = heartbeatManager.findStaleBrokers();
- for (int brokerId : staleBrokers) {
+ if (!staleBrokers.isEmpty()) {
+ // Even though multiple brokers can go stale at a time, we will process
+ // fencing one at a time so that the effect of fencing each broker is visible
+ // to the controller (memory) prior to processing the fencing of the next one
+ int brokerId = staleBrokers.get(0);
+ log.debug("Found broker(s) {} to be stale. Processing them one at a time",
+ Arrays.toString(staleBrokers.toArray()));
log.info("Fencing broker {} because its session has timed out.", brokerId);
handleBrokerFenced(brokerId, records);
heartbeatManager.fence(brokerId);
+
Review comment:
ack
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#issuecomment-897806958
@hachikuji Will follow up with a follow-up. Thanks for merging this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r686343189
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,112 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
Review comment:
The timeout was because I accidentally deleted a line before pushing the commit. Will fix that.
Will discuss pulling the test out offline with you.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
##########
@@ -858,7 +858,7 @@ private void rescheduleMaybeFenceStaleBrokers() {
return;
}
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs, () -> {
- ControllerResult<Void> result = replicationControl.maybeFenceStaleBrokers();
+ ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
rescheduleMaybeFenceStaleBrokers();
Review comment:
added tests in the next revision.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
##########
@@ -37,7 +42,13 @@
private final LocalLogManagerTestEnv logEnv;
public QuorumControllerTestEnv(LocalLogManagerTestEnv logEnv,
- Consumer<QuorumController.Builder> builderConsumer)
+ Consumer<QuorumController.Builder> builderConsumer)
Review comment:
ack. Will match the indentation to what it was before.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -925,18 +925,27 @@ ApiError electLeader(String topic, int partitionId, boolean uncleanOk,
return ControllerResult.of(records, null);
}
- ControllerResult<Void> maybeFenceStaleBrokers() {
+ ControllerResult<Void> maybeFenceOneStaleBroker() {
List<ApiMessageAndVersion> records = new ArrayList<>();
BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
- List<Integer> staleBrokers = heartbeatManager.findStaleBrokers();
- for (int brokerId : staleBrokers) {
+ Optional<Integer> staleBroker = heartbeatManager.findOneStaleBroker();
+ if (staleBroker.isPresent()) {
Review comment:
Will do.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/BrokerHeartbeatManagerTest.java
##########
@@ -93,22 +93,26 @@ public void testFindStaleBrokers() {
assertEquals(1, iter.next().id());
assertEquals(2, iter.next().id());
assertFalse(iter.hasNext());
- assertEquals(Collections.emptyList(), manager.findStaleBrokers());
+ assertEquals(Optional.empty(), manager.findOneStaleBroker());
time.sleep(5);
- assertEquals(Collections.singletonList(0), manager.findStaleBrokers());
+ assertEquals(Optional.of(0), manager.findOneStaleBroker());
manager.fence(0);
- assertEquals(Collections.emptyList(), manager.findStaleBrokers());
+ assertEquals(Optional.empty(), manager.findOneStaleBroker());
iter = manager.unfenced().iterator();
assertEquals(1, iter.next().id());
assertEquals(2, iter.next().id());
assertFalse(iter.hasNext());
time.sleep(20);
- assertEquals(Arrays.asList(1, 2), manager.findStaleBrokers());
- manager.fence(1);
- manager.fence(2);
- assertEquals(Collections.emptyList(), manager.findStaleBrokers());
+ Integer nodeId = 1;
+ while (manager.findOneStaleBroker().isPresent()) {
Review comment:
ack. Good catch! I should have added an assert on the node count.
Will just unroll the loop .
##########
File path: metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java
##########
@@ -419,6 +419,27 @@ long nextCheckTimeNs() {
}
}
+ /**
+ * Check if the oldest broker to have hearbeated has already violated the
+ * sessionTimeoutNs timeout and needs to be fenced.
+ *
+ * @return An Optional broker node id.
+ */
+ Optional<Integer> findOneStaleBroker() {
+ Optional<Integer> node = Optional.empty();
+ BrokerHeartbeatStateIterator iterator = unfenced.iterator();
+ if (iterator.hasNext()) {
+ BrokerHeartbeatState broker = iterator.next();
+ // The unfenced broker list is sorted on last contact time from each
+ // broker. If the first broker has a valid session then all do
+ if (!hasValidSession(broker)) {
+ node = Optional.of(broker.id);
Review comment:
will do.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -1029,6 +1051,22 @@ public void testCreatePartitionsFailsWithManualAssignmentWithAllFenced() throws
ctx.replicationControl.getPartition(fooId, 1));
}
+ @Test
+ public void testFenceMultipleBrokers() throws Exception {
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext();
+ ReplicationControlManager replication = ctx.replicationControl;
+ ctx.registerBrokers(0, 1, 2, 3, 4);
+ ctx.unfenceBrokers(0, 1, 2, 3, 4);
+
+ Uuid fooId = ctx.createTestTopic("foo", new int[][]{
+ new int[]{1, 2, 3}, new int[]{2, 3, 4}, new int[]{0, 2, 1}}).topicId();
+
+ ctx.fenceBrokers(Utils.mkSet(2, 3));
+
+ PartitionRegistration partition0 = replication.getPartition(fooId, 0);
+ assertArrayEquals(new int[]{1}, partition0.isr);
Review comment:
Makes sense. Will add the leader assertion and see what other things we can assert here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687262200
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
Review comment:
Will make the change. Had read the session timeout parameter wrong. Will keep consistent time units.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687262311
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
+ setValue("2")).iterator())).
+ setReplicationFactor(replicationFactor)).iterator()));
+ createTopicsResponseData = active.createTopics(createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("bar").errorCode());
+ Uuid topicIdBar = createTopicsResponseData.topics().find("bar").topicId();
+
+ // Fence some of the brokers
+ TestUtils.waitForCondition(() -> {
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ for (int i = brokersToKeepUnfenced; i < brokerCount; i++) {
+ if (active.replicationControl().isBrokerUnfenced(i)) {
+ return false;
+ }
+ }
+ return true;
+ }, sessionTimeoutSec * 2 * 1000,
+ "Fencing of brokers did not process within expected time"
+ );
+
+ // At this point the brokers we want fenced are fenced.
+ // Send another heartbeat to the brokers we want to keep alive
+ sendBrokerheartbeat(active, brokersToKeepUnfenced, brokerEpochs);
+ int[] expectedIsr = new int[brokersToKeepUnfenced];
Review comment:
Given we are keeping only one broker to remain unfenced, the expected isr and leader are simply that broker. I kept this logic to allow for paramterization of the test. Given we are not using the parameters right now, will remove this logic.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
##########
@@ -36,16 +41,29 @@
private final List<QuorumController> controllers;
private final LocalLogManagerTestEnv logEnv;
- public QuorumControllerTestEnv(LocalLogManagerTestEnv logEnv,
- Consumer<QuorumController.Builder> builderConsumer)
- throws Exception {
+ public QuorumControllerTestEnv(
+ LocalLogManagerTestEnv logEnv,
+ Consumer<QuorumController.Builder> builderConsumer
+ ) throws Exception {
+ this(logEnv, builderConsumer, Optional.empty());
+ }
+
+ public QuorumControllerTestEnv(
+ LocalLogManagerTestEnv logEnv,
+ Consumer<Builder> builderConsumer,
+ Optional<Long> sessionTimeoutSeconds
Review comment:
Will do.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r687262147
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
+ Long sleepMillis = (sessionTimeoutSec * 1000) / 2;
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv =
+ new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS), Optional.of(sessionTimeoutSec))) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ new BrokerRegistrationRequestData().
+ setBrokerId(brokerId).
+ setClusterId("06B-K3N1TBCNYFgruEVP0Q").
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setListeners(listeners));
+ brokerEpochs.put(brokerId, reply.get().epoch());
+ }
+
+ // Brokers are only registered but still fenced
+ // Topic creation with no available unfenced brokers should fail
+ CreateTopicsRequestData createTopicsRequestData =
+ new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("foo").setNumPartitions(1).
+ setReplicationFactor(replicationFactor)).iterator()));
+ CreateTopicsResponseData createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.INVALID_REPLICATION_FACTOR.code(),
+ createTopicsResponseData.topics().find("foo").errorCode());
+ assertEquals("Unable to replicate the partition " + replicationFactor + " time(s): All brokers " +
+ "are currently fenced.", createTopicsResponseData.topics().find("foo").errorMessage());
+
+ // Unfence all brokers
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ createTopicsResponseData = active.createTopics(
+ createTopicsRequestData).get();
+ assertEquals(Errors.NONE.code(), createTopicsResponseData.topics().find("foo").errorCode());
+ Uuid topicIdFoo = createTopicsResponseData.topics().find("foo").topicId();
+ sendBrokerheartbeat(active, brokerCount, brokerEpochs);
+ Thread.sleep(sleepMillis);
+
+ // All brokers should still be unfenced
+ for (int brokerId = 0; brokerId < brokerCount; brokerId++) {
+ assertTrue(active.replicationControl().isBrokerUnfenced(brokerId),
+ "Broker " + brokerId + " should have been unfenced");
+ }
+ createTopicsRequestData = new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName("bar").setNumPartitions(1).
+ setConfigs(new CreateableTopicConfigCollection(Collections.
+ singleton(new CreateableTopicConfig().setName("min.insync.replicas").
Review comment:
TL;DR Is vestigial, will remove.
The reason i added it in there was because even with only one broker available the topic creation would succeed. This added condition of a min.isr was to strengthen the topic creation criteria to make it fail when enough brokers were not un-fenced. I learnt that today the behavior is to not block topic creation even if the min.isr would not be met for the topic on a subsequent produce. Anyway, unless that behavior is changed the min.isr is serving no purpose and I will remove it. I added explicit checks to ensure broker liveness.
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,121 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
+ int brokerCount = 5;
+ int brokersToKeepUnfenced = 1;
+ short replicationFactor = 5;
+ Long sessionTimeoutSec = 1L;
Review comment:
Will make the change. The issue is that the sleepTimeout configuration takes in seconds and I wanted to skip the coversion there. But I will make the time unit consistent across the test.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r686135001
##########
File path: metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java
##########
@@ -419,6 +419,27 @@ long nextCheckTimeNs() {
}
}
+ /**
+ * Check if the oldest broker to have hearbeated has already violated the
+ * sessionTimeoutNs timeout and needs to be fenced.
+ *
+ * @return An Optional broker node id.
+ */
+ Optional<Integer> findOneStaleBroker() {
+ Optional<Integer> node = Optional.empty();
+ BrokerHeartbeatStateIterator iterator = unfenced.iterator();
+ if (iterator.hasNext()) {
+ BrokerHeartbeatState broker = iterator.next();
+ // The unfenced broker list is sorted on last contact time from each
+ // broker. If the first broker has a valid session then all do
+ if (!hasValidSession(broker)) {
+ node = Optional.of(broker.id);
Review comment:
nit: you can return here and then we don't need `node`
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -925,18 +925,27 @@ ApiError electLeader(String topic, int partitionId, boolean uncleanOk,
return ControllerResult.of(records, null);
}
- ControllerResult<Void> maybeFenceStaleBrokers() {
+ ControllerResult<Void> maybeFenceOneStaleBroker() {
List<ApiMessageAndVersion> records = new ArrayList<>();
BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
- List<Integer> staleBrokers = heartbeatManager.findStaleBrokers();
- for (int brokerId : staleBrokers) {
+ Optional<Integer> staleBroker = heartbeatManager.findOneStaleBroker();
+ if (staleBroker.isPresent()) {
Review comment:
nit: maybe a little more concise with a lambda?
```java
heartbeatManager.findOneStaleBroker().ifPresent(brokerId -> {
// Even though multiple brokers can go stale at a time, we will process
// fencing one at a time so that the effect of fencing each broker is visible
// to the system prior to processing the next one
log.info("Fencing broker {} because its session has timed out.", brokerId);
handleBrokerFenced(brokerId, records);
heartbeatManager.fence(brokerId);
});
```
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
##########
@@ -173,6 +176,112 @@ private void testDelayedConfigurationOperations(LocalLogManagerTestEnv logEnv,
assertEquals(Collections.singletonMap(BROKER0, ApiError.NONE), future1.get());
}
+ @Test
+ public void testFenceMultipleBrokers() throws Throwable {
Review comment:
This test times out for me when run locally. I'm a little concerned that the dependence on real time here will make it flaky. I'm satisfied with the other tests we have in this patch. Would it be reasonable to push this test case to a follow-up so that we can iterate on it a little bit without blocking the patch?
##########
File path: metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
##########
@@ -37,7 +42,13 @@
private final LocalLogManagerTestEnv logEnv;
public QuorumControllerTestEnv(LocalLogManagerTestEnv logEnv,
- Consumer<QuorumController.Builder> builderConsumer)
+ Consumer<QuorumController.Builder> builderConsumer)
Review comment:
nit: the indentation is a bit screwy here and below
##########
File path: metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
##########
@@ -1029,6 +1051,22 @@ public void testCreatePartitionsFailsWithManualAssignmentWithAllFenced() throws
ctx.replicationControl.getPartition(fooId, 1));
}
+ @Test
+ public void testFenceMultipleBrokers() throws Exception {
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext();
+ ReplicationControlManager replication = ctx.replicationControl;
+ ctx.registerBrokers(0, 1, 2, 3, 4);
+ ctx.unfenceBrokers(0, 1, 2, 3, 4);
+
+ Uuid fooId = ctx.createTestTopic("foo", new int[][]{
+ new int[]{1, 2, 3}, new int[]{2, 3, 4}, new int[]{0, 2, 1}}).topicId();
+
+ ctx.fenceBrokers(Utils.mkSet(2, 3));
+
+ PartitionRegistration partition0 = replication.getPartition(fooId, 0);
+ assertArrayEquals(new int[]{1}, partition0.isr);
Review comment:
Can we extend this test to assert something about the other partitions? Maybe also useful to assert the current leader?
##########
File path: metadata/src/test/java/org/apache/kafka/controller/BrokerHeartbeatManagerTest.java
##########
@@ -93,22 +93,26 @@ public void testFindStaleBrokers() {
assertEquals(1, iter.next().id());
assertEquals(2, iter.next().id());
assertFalse(iter.hasNext());
- assertEquals(Collections.emptyList(), manager.findStaleBrokers());
+ assertEquals(Optional.empty(), manager.findOneStaleBroker());
time.sleep(5);
- assertEquals(Collections.singletonList(0), manager.findStaleBrokers());
+ assertEquals(Optional.of(0), manager.findOneStaleBroker());
manager.fence(0);
- assertEquals(Collections.emptyList(), manager.findStaleBrokers());
+ assertEquals(Optional.empty(), manager.findOneStaleBroker());
iter = manager.unfenced().iterator();
assertEquals(1, iter.next().id());
assertEquals(2, iter.next().id());
assertFalse(iter.hasNext());
time.sleep(20);
- assertEquals(Arrays.asList(1, 2), manager.findStaleBrokers());
- manager.fence(1);
- manager.fence(2);
- assertEquals(Collections.emptyList(), manager.findStaleBrokers());
+ Integer nodeId = 1;
+ while (manager.findOneStaleBroker().isPresent()) {
Review comment:
nit: maybe it's better to unroll this to make the expectation explicit. This test would pass even if there were additional iterations of the loop.
```java
assertEquals(Optional.of(1), manager.findStaleBroker());
manager.fence(1);
assertEquals(Optional.of(2), manager.findStaleBroker());
manager.fence(2);
```
Ditto below.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a change in pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
niket-goel commented on a change in pull request #11191:
URL: https://github.com/apache/kafka/pull/11191#discussion_r685437960
##########
File path: metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
##########
@@ -858,7 +858,7 @@ private void rescheduleMaybeFenceStaleBrokers() {
return;
}
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs, () -> {
- ControllerResult<Void> result = replicationControl.maybeFenceStaleBrokers();
+ ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
rescheduleMaybeFenceStaleBrokers();
Review comment:
I thought about doing that, but then realized that a call to reschedule will automatically result in a scheduling of another invocation right away if there were any brokers that were deemed stale still remaining in the unfenced list.
This will happen because the `nextCheckTimeNs` is based on when the broker with the older `lastContactNs` time (broker at the head of the list) would time out. This will always result in a time less than now if we have a stale broker and there should be no delay in executing this again (barring queue depth).
I also did not want to "prioritize" (if that's even possible) over other requests, because my understanding is that the fix works as long as a subsequent fence broker call is made after the first has been processed by the controller thread (which would also happen by picking up the record from the queue and replaying it).
Please let me know if my understanding is off here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #11191: KAFKA-13173 Making the controller fence stale brokers one at a time if multiple stale brokers are discovered at the same time
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #11191:
URL: https://github.com/apache/kafka/pull/11191
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org