You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2022/06/23 15:29:33 UTC
[kafka] branch trunk updated: KAFKA-13966 Prepend bootstrap metadata to controller queue (#12269)
This is an automated email from the ASF dual-hosted git repository.
davidarthur pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new c6c9da02a8 KAFKA-13966 Prepend bootstrap metadata to controller queue (#12269)
c6c9da02a8 is described below
commit c6c9da02a81fcc034ea916230ac8acea603d41b7
Author: David Arthur <mu...@gmail.com>
AuthorDate: Thu Jun 23 11:29:21 2022 -0400
KAFKA-13966 Prepend bootstrap metadata to controller queue (#12269)
Also fixes flaky QuorumControllerTest#testInvalidBootstrapMetadata
Reviewers: Jason Gustafson <ja...@confluent.io>
---
.../apache/kafka/controller/QuorumController.java | 19 ++++-
.../kafka/controller/QuorumControllerTest.java | 86 ++++++++++++++++++++--
.../kafka/controller/QuorumControllerTestEnv.java | 10 +++
3 files changed, 105 insertions(+), 10 deletions(-)
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 92f08a67f5..0290e0040c 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -782,6 +782,13 @@ public final class QuorumController implements Controller {
}
}
+ private <T> CompletableFuture<T> prependWriteEvent(String name,
+ ControllerWriteOperation<T> op) {
+ ControllerWriteEvent<T> event = new ControllerWriteEvent<>(name, op);
+ queue.prepend(event);
+ return event.future();
+ }
+
private <T> CompletableFuture<T> appendWriteEvent(String name,
OptionalLong deadlineNs,
ControllerWriteOperation<T> op) {
@@ -922,7 +929,6 @@ public final class QuorumController implements Controller {
curEpoch);
}
-
curClaimEpoch = newEpoch;
controllerMetrics.setActive(true);
updateWriteOffset(lastCommittedOffset);
@@ -942,14 +948,20 @@ public final class QuorumController implements Controller {
"Got " + bootstrapMetadata.metadataVersion()));
} else {
metadataVersion = bootstrapMetadata.metadataVersion();
- future = appendWriteEvent("bootstrapMetadata", OptionalLong.empty(), () -> {
+
+ // This call is here instead of inside the appendWriteEvent for testing purposes.
+ final List<ApiMessageAndVersion> bootstrapRecords = bootstrapMetadata.records();
+
+ // We prepend the bootstrap event in order to ensure the bootstrap metadata is written before
+ // any external controller write events are processed.
+ future = prependWriteEvent("bootstrapMetadata", () -> {
if (metadataVersion.isAtLeast(MetadataVersion.IBP_3_3_IV0)) {
log.info("Initializing metadata.version to {}", metadataVersion.featureLevel());
} else {
log.info("Upgrading KRaft cluster and initializing metadata.version to {}",
metadataVersion.featureLevel());
}
- return ControllerResult.atomicOf(bootstrapMetadata.records(), null);
+ return ControllerResult.atomicOf(bootstrapRecords, null);
});
}
future.whenComplete((result, exception) -> {
@@ -967,7 +979,6 @@ public final class QuorumController implements Controller {
metadataVersion = featureControl.metadataVersion();
}
-
log.info(
"Becoming the active controller at epoch {}, committed offset {}, committed epoch {}, and metadata.version {}",
newEpoch, lastCommittedOffset, lastCommittedEpoch, metadataVersion.featureLevel()
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
index 34ad4d9727..a62b1f682f 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -32,6 +32,7 @@ import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -536,11 +537,16 @@ public class QuorumControllerTest {
}
private BrokerRegistrationRequestData.FeatureCollection brokerFeatures() {
+ return brokerFeatures(MetadataVersion.MINIMUM_KRAFT_VERSION, MetadataVersion.latest());
+ }
+
+ private BrokerRegistrationRequestData.FeatureCollection brokerFeatures(
+ MetadataVersion minVersion, MetadataVersion maxVersion) {
BrokerRegistrationRequestData.FeatureCollection features = new BrokerRegistrationRequestData.FeatureCollection();
features.add(new BrokerRegistrationRequestData.Feature()
.setName(MetadataVersion.FEATURE_NAME)
- .setMinSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel())
- .setMaxSupportedVersion(MetadataVersion.latest().featureLevel()));
+ .setMinSupportedVersion(minVersion.featureLevel())
+ .setMaxSupportedVersion(maxVersion.featureLevel()));
return features;
}
@@ -1184,18 +1190,86 @@ public class QuorumControllerTest {
@Test
public void testInvalidBootstrapMetadata() throws Exception {
- // We can't actually create a BootstrapMetadata with an invalid version, so we have to fake it
+ // We can't actually create a BootstrapMetadata with an invalid version, so we have to mock it
BootstrapMetadata bootstrapMetadata = Mockito.mock(BootstrapMetadata.class);
- Mockito.when(bootstrapMetadata.metadataVersion()).thenReturn(MetadataVersion.IBP_2_8_IV0);
+ CyclicBarrier barrier = new CyclicBarrier(2);
+ Mockito.when(bootstrapMetadata.metadataVersion()).thenAnswer(__ -> {
+ // This barrier allows us to catch the controller after it becomes leader, but before the bootstrapping fails
+ barrier.await(10, TimeUnit.SECONDS);
+ return MetadataVersion.IBP_2_8_IV0;
+ });
try (
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty());
QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv(logEnv, b -> {
b.setConfigSchema(SCHEMA);
}, OptionalLong.empty(), OptionalLong.empty(), bootstrapMetadata);
) {
- QuorumController active = controlEnv.activeController();
- TestUtils.waitForCondition(() -> !active.isActive(),
+ QuorumController controller = controlEnv.activeController();
+ assertTrue(controller.isActive());
+ // Unblock the first call to BootstrapMetadata#metadataVersion
+ barrier.await(10, TimeUnit.SECONDS);
+ // Unblock the second call to BootstrapMetadata#metadataVersion
+ barrier.await(10, TimeUnit.SECONDS);
+ TestUtils.waitForCondition(() -> !controller.isActive(),
"Timed out waiting for controller to renounce itself after bad bootstrap metadata version.");
}
}
+
+ @Test
+ public void testBootstrapMetadataStartupRace() throws Throwable {
+ // KAFKA-13966: This tests a race condition between external RPC calls being handled before the bootstrap
+ // metadata is written. We instrument this by forcing the BootstrapMetadata#records method to block until a
+ // latch has been completed. This allows an asynchronous broker registration call to be handled before the
+ // handleLeaderChange callback completes. In this case, the registration should fail because the bootstrap
+ // metadata includes an unsupported metadata.version.
+ BootstrapMetadata bootstrapMetadata = BootstrapMetadata.create(MetadataVersion.latest());
+ BootstrapMetadata mockedMetadata = Mockito.mock(BootstrapMetadata.class);
+ CountDownLatch latch = new CountDownLatch(1);
+ Mockito.when(mockedMetadata.metadataVersion()).thenReturn(bootstrapMetadata.metadataVersion());
+ Mockito.when(mockedMetadata.records()).then(__ -> {
+ if (latch.await(30, TimeUnit.SECONDS)) {
+ return bootstrapMetadata.records();
+ } else {
+ throw new RuntimeException("Latch never completed");
+ }
+ });
+
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(1, Optional.empty())) {
+ try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv(logEnv, b -> {
+ b.setConfigSchema(SCHEMA);
+ }, OptionalLong.empty(), OptionalLong.empty(), mockedMetadata)) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+
+ // Issue a register broker request concurrently as the controller is initializing
+ assertEquals(1, latch.getCount(), "Latch should not have been completed yet");
+ CompletableFuture<Void> registrationFuture = new CompletableFuture<>();
+ Thread registerThread = new Thread(() -> {
+ try {
+ CompletableFuture<BrokerRegistrationReply> reply = active.registerBroker(
+ ANONYMOUS_CONTEXT,
+ new BrokerRegistrationRequestData().
+ setBrokerId(0).
+ setClusterId(active.clusterId()).
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV0, MetadataVersion.IBP_3_3_IV0)).
+ setListeners(listeners));
+ // Once we have the future, the register broker event has been enqueued
+ latch.countDown();
+ reply.get();
+ registrationFuture.complete(null);
+ } catch (Throwable t) {
+ registrationFuture.completeExceptionally(t);
+ }
+ });
+ registerThread.start();
+ registerThread.join(30_000);
+ assertTrue(registrationFuture.isCompletedExceptionally(),
+ "Should not be able to register broker since the bootstrap metadata specified an incompatible metadata.version");
+ assertEquals(0, active.clusterControl().brokerRegistrations().size());
+ }
+ }
+ }
}
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
index 68b11ae3ec..4cc45a9774 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
@@ -52,6 +52,16 @@ public class QuorumControllerTestEnv implements AutoCloseable {
this(logEnv, builderConsumer, OptionalLong.empty(), OptionalLong.empty(), BootstrapMetadata.create(MetadataVersion.latest()));
}
+ public QuorumControllerTestEnv(
+ LocalLogManagerTestEnv logEnv,
+ Consumer<Builder> builderConsumer,
+ OptionalLong sessionTimeoutMillis,
+ OptionalLong leaderImbalanceCheckIntervalNs,
+ MetadataVersion metadataVersion
+ ) throws Exception {
+ this(logEnv, builderConsumer, sessionTimeoutMillis, leaderImbalanceCheckIntervalNs, BootstrapMetadata.create(metadataVersion));
+ }
+
public QuorumControllerTestEnv(
LocalLogManagerTestEnv logEnv,
Consumer<Builder> builderConsumer,