You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2024/08/21 18:15:28 UTC
(kafka) branch 3.9 updated: KAFKA-17305; Check broker registrations for missing features (#16848)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new b377ea94c5f KAFKA-17305; Check broker registrations for missing features (#16848)
b377ea94c5f is described below
commit b377ea94c5fae39bb3b06f203de8c95afbd7a423
Author: Alyssa Huang <ah...@confluent.io>
AuthorDate: Wed Aug 21 11:14:56 2024 -0700
KAFKA-17305; Check broker registrations for missing features (#16848)
When a broker tries to register with the controller quorum, its registration should be rejected if it doesn't support a feature that is currently enabled. (A feature is enabled if it is set to a non-zero feature level.) This is important for the newly added kraft.version feature flag.
Reviewers: Colin P. McCabe <cm...@apache.org>, José Armando García Sancio <js...@apache.org>
---
.../main/scala/kafka/server/BrokerFeatures.scala | 8 ++
.../src/main/scala/kafka/server/BrokerServer.scala | 8 +-
core/src/main/scala/kafka/server/KafkaServer.scala | 11 +-
.../kafka/server/BrokerLifecycleManagerTest.scala | 53 +++++++-
.../server/BrokerRegistrationRequestTest.scala | 40 +++++-
.../unit/kafka/server/ReplicationQuotasTest.scala | 12 +-
.../kafka/controller/ClusterControlManager.java | 41 ++++--
.../apache/kafka/controller/QuorumController.java | 8 +-
.../controller/ClusterControlManagerTest.java | 142 ++++++++++++++++++++-
.../kafka/controller/QuorumControllerTest.java | 59 +++++++++
.../org/apache/kafka/metalog/LocalLogManager.java | 11 +-
.../kafka/metalog/LocalLogManagerTestEnv.java | 19 ++-
12 files changed, 371 insertions(+), 41 deletions(-)
diff --git a/core/src/main/scala/kafka/server/BrokerFeatures.scala b/core/src/main/scala/kafka/server/BrokerFeatures.scala
index 5aab78d4a2a..6a8ddcb19c7 100644
--- a/core/src/main/scala/kafka/server/BrokerFeatures.scala
+++ b/core/src/main/scala/kafka/server/BrokerFeatures.scala
@@ -19,6 +19,7 @@ package kafka.server
import kafka.utils.Logging
import org.apache.kafka.common.feature.{Features, SupportedVersionRange}
+import org.apache.kafka.metadata.VersionRange
import org.apache.kafka.server.common.Features.PRODUCTION_FEATURES
import org.apache.kafka.server.common.MetadataVersion
@@ -80,6 +81,13 @@ object BrokerFeatures extends Logging {
new BrokerFeatures(defaultSupportedFeatures(unstableFeatureVersionsEnabled))
}
+ def createDefaultFeatureMap(features: BrokerFeatures): Map[String, VersionRange] = {
+ features.supportedFeatures.features.asScala.map {
+ case (name, versionRange) =>
+ (name, VersionRange.of(versionRange.min, versionRange.max))
+ }.toMap
+ }
+
def defaultSupportedFeatures(unstableFeatureVersionsEnabled: Boolean): Features[SupportedVersionRange] = {
val features = new util.HashMap[String, SupportedVersionRange]()
features.put(MetadataVersion.FEATURE_NAME,
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index 790bc4ccd1d..113bd87989a 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -27,7 +27,6 @@ import kafka.raft.KafkaRaftManager
import kafka.server.metadata.{AclPublisher, BrokerMetadataPublisher, ClientQuotaMetadataManager, DelegationTokenPublisher, DynamicClientQuotaPublisher, DynamicConfigPublisher, KRaftMetadataCache, ScramPublisher}
import kafka.utils.CoreUtils
import org.apache.kafka.common.config.ConfigException
-import org.apache.kafka.common.feature.SupportedVersionRange
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.ListenerName
@@ -38,7 +37,7 @@ import org.apache.kafka.common.{ClusterResource, TopicPartition, Uuid}
import org.apache.kafka.coordinator.group.metrics.{GroupCoordinatorMetrics, GroupCoordinatorRuntimeMetrics}
import org.apache.kafka.coordinator.group.{CoordinatorRecord, GroupCoordinator, GroupCoordinatorService, CoordinatorRecordSerde}
import org.apache.kafka.image.publisher.{BrokerRegistrationTracker, MetadataPublisher}
-import org.apache.kafka.metadata.{BrokerState, ListenerInfo, VersionRange}
+import org.apache.kafka.metadata.{BrokerState, ListenerInfo}
import org.apache.kafka.security.CredentialProvider
import org.apache.kafka.server.{AssignmentsManager, ClientMetricsManager, NodeToControllerChannelManager}
import org.apache.kafka.server.authorizer.Authorizer
@@ -357,10 +356,7 @@ class BrokerServer(
ConfigType.BROKER -> new BrokerConfigHandler(config, quotaManagers),
ConfigType.CLIENT_METRICS -> new ClientMetricsConfigHandler(clientMetricsManager))
- val featuresRemapped = brokerFeatures.supportedFeatures.features().asScala.map {
- case (k: String, v: SupportedVersionRange) =>
- k -> VersionRange.of(v.min, v.max)
- }.asJava
+ val featuresRemapped = BrokerFeatures.createDefaultFeatureMap(brokerFeatures).asJava
val brokerLifecycleChannelManager = new NodeToControllerChannelManagerImpl(
controllerNodeProvider,
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 78ddc408d6b..4b1dec32126 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -474,18 +474,19 @@ class KafkaServer(
setSecurityProtocol(ep.securityProtocol.id))
}
- // Even though ZK brokers don't use "metadata.version" feature, we send our IBP here as part of the broker registration
+ val features = BrokerFeatures.createDefaultFeatureMap(BrokerFeatures.createDefault(config.unstableFeatureVersionsEnabled))
+
+ // Even though ZK brokers don't use "metadata.version" feature, we need to overwrite it with our IBP as part of registration
// so the KRaft controller can verify that all brokers are on the same IBP before starting the migration.
- val ibpAsFeature =
- java.util.Collections.singletonMap(MetadataVersion.FEATURE_NAME,
- VersionRange.of(config.interBrokerProtocolVersion.featureLevel(), config.interBrokerProtocolVersion.featureLevel()))
+ val featuresRemapped = features + (MetadataVersion.FEATURE_NAME ->
+ VersionRange.of(config.interBrokerProtocolVersion.featureLevel(), config.interBrokerProtocolVersion.featureLevel()))
lifecycleManager.start(
() => listener.highestOffset,
brokerToQuorumChannelManager,
clusterId,
networkListeners,
- ibpAsFeature,
+ featuresRemapped.asJava,
OptionalLong.empty()
)
logger.debug("Start RaftManager")
diff --git a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
index 71bfbefa307..8e88eea3c46 100644
--- a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
@@ -24,11 +24,15 @@ import org.apache.kafka.common.Uuid
import org.apache.kafka.common.message.{BrokerHeartbeatResponseData, BrokerRegistrationResponseData}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
-import org.apache.kafka.metadata.BrokerState
+import org.apache.kafka.metadata.{BrokerState, VersionRange}
import org.apache.kafka.raft.QuorumConfig
-import org.apache.kafka.server.config.{KRaftConfigs, ServerLogConfigs}
+import org.apache.kafka.server.common.{Features, KRaftVersion, MetadataVersion}
+import org.apache.kafka.server.common.MetadataVersion.{IBP_3_8_IV0, IBP_3_9_IV0}
+import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs, ServerLogConfigs, ZkConfigs}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test, Timeout}
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import java.util.concurrent.{CompletableFuture, Future}
import scala.jdk.CollectionConverters._
@@ -56,6 +60,15 @@ class BrokerLifecycleManagerTest {
properties
}
+ def migrationConfigProperties(ibp: MetadataVersion) = {
+ val migrationConfigProperties = configProperties
+ migrationConfigProperties.setProperty(KRaftConfigs.MIGRATION_ENABLED_CONFIG, "true")
+ migrationConfigProperties.setProperty(ZkConfigs.ZK_CONNECT_CONFIG, "localhost:2181")
+ migrationConfigProperties.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "")
+ migrationConfigProperties.setProperty(ReplicationConfigs.INTER_BROKER_PROTOCOL_VERSION_CONFIG, ibp.toString)
+ migrationConfigProperties
+ }
+
@Test
def testCreateAndClose(): Unit = {
val context = new RegistrationTestContext(configProperties)
@@ -99,6 +112,42 @@ class BrokerLifecycleManagerTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def testSuccessfulRegistrationDuringMigration(nonInitialKraftVersion: Boolean): Unit = {
+ val ibp = if (nonInitialKraftVersion) IBP_3_9_IV0 else IBP_3_8_IV0
+ val context = new RegistrationTestContext(migrationConfigProperties(ibp))
+ manager = new BrokerLifecycleManager(context.config, context.time, "successful-registration-", isZkBroker = false, Set(Uuid.fromString("gCpDJgRlS2CBCpxoP2VMsQ")))
+ val controllerNode = new Node(3000, "localhost", 8021)
+ context.controllerNodeProvider.node.set(controllerNode)
+ val features = BrokerFeatures.createDefaultFeatureMap(BrokerFeatures.createDefault(true))
+
+ // Even though ZK brokers don't use "metadata.version" feature, we need to overwrite it with our IBP as part of registration
+ // so the KRaft controller can verify that all brokers are on the same IBP before starting the migration.
+ val featuresRemapped = features + (MetadataVersion.FEATURE_NAME -> VersionRange.of(ibp.featureLevel(), ibp.featureLevel()))
+
+ manager.start(() => context.highestMetadataOffset.get(),
+ context.mockChannelManager, context.clusterId, context.advertisedListeners,
+ featuresRemapped.asJava, OptionalLong.of(10L))
+ TestUtils.retry(60000) {
+ assertEquals(1, context.mockChannelManager.unsentQueue.size)
+ val sentBrokerRegistrationData = context.mockChannelManager.unsentQueue.getFirst.request.build().asInstanceOf[BrokerRegistrationRequest].data()
+ assertEquals(10L, sentBrokerRegistrationData.previousBrokerEpoch())
+ assertEquals(ibp.featureLevel(), sentBrokerRegistrationData.features().find(MetadataVersion.FEATURE_NAME).maxSupportedVersion())
+ if (nonInitialKraftVersion) {
+ val sentKraftVersion = sentBrokerRegistrationData.features().find(KRaftVersion.FEATURE_NAME)
+ assertEquals(Features.KRAFT_VERSION.minimumProduction(), sentKraftVersion.minSupportedVersion())
+ assertEquals(Features.KRAFT_VERSION.latestTesting(), sentKraftVersion.maxSupportedVersion())
+ }
+ }
+ context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
+ new BrokerRegistrationResponseData().setBrokerEpoch(1000)), controllerNode)
+ TestUtils.retry(10000) {
+ context.poll()
+ assertEquals(1000L, manager.brokerEpoch)
+ }
+ }
+
@Test
def testRegistrationTimeout(): Unit = {
val context = new RegistrationTestContext(configProperties)
diff --git a/core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala b/core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala
index 247973255b2..f60aac80c90 100644
--- a/core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala
@@ -32,11 +32,12 @@ import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.utils.Time
import org.apache.kafka.common.{Node, Uuid}
import org.apache.kafka.server.{ControllerRequestCompletionHandler, NodeToControllerChannelManager}
-import org.apache.kafka.server.common.MetadataVersion
+import org.apache.kafka.server.common.{Features, MetadataVersion}
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
import org.junit.jupiter.api.Timeout
import org.junit.jupiter.api.extension.ExtendWith
+import java.util.Collections
import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}
/**
@@ -98,7 +99,6 @@ class BrokerRegistrationRequestTest {
ibpToSend: Option[(MetadataVersion, MetadataVersion)]
): Errors = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
-
ibpToSend foreach {
case (min, max) =>
features.add(new BrokerRegistrationRequestData.Feature()
@@ -107,9 +107,17 @@ class BrokerRegistrationRequestTest {
.setMaxSupportedVersion(max.featureLevel())
)
}
+ Features.PRODUCTION_FEATURES.stream().filter(_.featureName != MetadataVersion.FEATURE_NAME).forEach {
+ feature =>
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName)
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting()))
+ }
val req = new BrokerRegistrationRequestData()
.setBrokerId(brokerId)
+ .setLogDirs(Collections.singletonList(Uuid.randomUuid()))
.setClusterId(clusterId)
.setIncarnationId(Uuid.randomUuid())
.setIsMigratingZkBroker(zkEpoch.isDefined)
@@ -229,6 +237,34 @@ class BrokerRegistrationRequestTest {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ brokers = 1,
+ controllers = 1,
+ metadataVersion = MetadataVersion.IBP_3_9_IV0,
+ autoStart = AutoStart.NO,
+ serverProperties = Array(new ClusterConfigProperty(key = "zookeeper.metadata.migration.enable", value = "true")))
+ def testRegisterZkWithKRaftMigrationEnabledKRaftV1(clusterInstance: ClusterInstance): Unit = {
+ clusterInstance.asInstanceOf[RaftClusterInstance].controllers().values().forEach(_.startup())
+
+ val clusterId = clusterInstance.clusterId()
+ val channelManager = brokerToControllerChannelManager(clusterInstance)
+ try {
+ channelManager.start()
+
+ assertEquals(
+ Errors.NONE,
+ registerBroker(channelManager, clusterId, 100, Some(1), Some((MetadataVersion.IBP_3_9_IV0, MetadataVersion.IBP_3_9_IV0))))
+
+ // Cannot register KRaft broker when in pre-migration
+ assertEquals(
+ Errors.BROKER_ID_NOT_REGISTERED,
+ registerBroker(channelManager, clusterId, 100, None, Some((MetadataVersion.IBP_3_9_IV0, MetadataVersion.IBP_3_9_IV0))))
+ } finally {
+ channelManager.shutdown()
+ }
+ }
+
/**
* Start a KRaft cluster with migrations enabled, verify that the controller does not accept metadata changes
* through the RPCs. The migration never proceeds past pre-migration since no ZK brokers are registered.
diff --git a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
index 3353c6efcc5..ff697dad370 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
@@ -35,7 +35,7 @@ import org.apache.kafka.common.message.BrokerRegistrationRequestData.{Listener,
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.SecurityProtocol.PLAINTEXT
import org.apache.kafka.controller.ControllerRequestContextUtil
-import org.apache.kafka.server.common.MetadataVersion
+import org.apache.kafka.server.common.{Features, KRaftVersion, MetadataVersion, TransactionVersion}
import org.apache.kafka.server.config.QuotaConfigs
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.AfterEach
@@ -291,10 +291,18 @@ class ReplicationQuotasTest extends QuorumTestHarness {
val listeners = new ListenerCollection()
listeners.add(new Listener().setName(PLAINTEXT.name).setHost("localhost").setPort(9092 + id))
val features = new BrokerRegistrationRequestData.FeatureCollection()
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(KRaftVersion.FEATURE_NAME)
+ .setMinSupportedVersion(KRaftVersion.KRAFT_VERSION_0.featureLevel())
+ .setMaxSupportedVersion(Features.KRAFT_VERSION.latestTesting()))
features.add(new BrokerRegistrationRequestData.Feature()
.setName(MetadataVersion.FEATURE_NAME)
- .setMinSupportedVersion(MetadataVersion.IBP_3_0_IV1.featureLevel())
+ .setMinSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel())
.setMaxSupportedVersion(MetadataVersion.latestTesting().featureLevel()))
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(TransactionVersion.FEATURE_NAME)
+ .setMinSupportedVersion(TransactionVersion.TV_0.featureLevel())
+ .setMaxSupportedVersion(Features.TRANSACTION_VERSION.latestTesting()))
controllerServer.controller.registerBroker(
ControllerRequestContextUtil.ANONYMOUS_CONTEXT,
new BrokerRegistrationRequestData()
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
index 7f77455abcc..77ffd1036be 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
@@ -50,7 +50,6 @@ import org.apache.kafka.metadata.placement.ReplicaPlacer;
import org.apache.kafka.metadata.placement.StripedReplicaPlacer;
import org.apache.kafka.metadata.placement.UsableBroker;
import org.apache.kafka.server.common.ApiMessageAndVersion;
-import org.apache.kafka.server.common.Features;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineHashMap;
@@ -406,18 +405,34 @@ public class ClusterControlManager {
setRack(request.rack()).
setEndPoints(listenerInfo.toBrokerRegistrationRecord());
+ // Track which finalized features we have not yet verified are supported by the broker.
+ Map<String, Short> unverifiedFeatures = new HashMap<>(finalizedFeatures.featureMap());
+
+ // Check every broker feature version range includes the finalized version.
for (BrokerRegistrationRequestData.Feature feature : request.features()) {
record.features().add(processRegistrationFeature(brokerId, finalizedFeatures, feature));
+ unverifiedFeatures.remove(feature.name());
}
+ // Brokers that don't send a supported metadata.version range are assumed to only
+ // support the original metadata.version.
if (request.features().find(MetadataVersion.FEATURE_NAME) == null) {
- // Brokers that don't send a supported metadata.version range are assumed to only
- // support the original metadata.version.
record.features().add(processRegistrationFeature(brokerId, finalizedFeatures,
+ new BrokerRegistrationRequestData.Feature().
+ setName(MetadataVersion.FEATURE_NAME).
+ setMinSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel())));
+ unverifiedFeatures.remove(MetadataVersion.FEATURE_NAME);
+ }
+ // We also need to check every controller feature is supported by the broker.
+ unverifiedFeatures.forEach((featureName, finalizedVersion) -> {
+ if (finalizedVersion != 0 && request.features().findAll(featureName).isEmpty()) {
+ processRegistrationFeature(brokerId, finalizedFeatures,
new BrokerRegistrationRequestData.Feature().
- setName(MetadataVersion.FEATURE_NAME).
- setMinSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel()).
- setMaxSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel())));
- }
+ setName(featureName).
+ setMinSupportedVersion((short) 0).
+ setMaxSupportedVersion((short) 0));
+ }
+ });
if (featureControl.metadataVersion().isDirectoryAssignmentSupported()) {
record.setLogDirs(request.logDirs());
}
@@ -490,15 +505,15 @@ public class ClusterControlManager {
short finalized = finalizedFeatures.versionOrDefault(feature.name(), (short) defaultVersion);
if (!VersionRange.of(feature.minSupportedVersion(), feature.maxSupportedVersion()).contains(finalized)) {
throw new UnsupportedVersionException("Unable to register because the broker " +
- "does not support version " + finalized + " of " + feature.name() +
- ". It wants a version between " + feature.minSupportedVersion() + " and " +
+ "does not support finalized version " + finalized + " of " + feature.name() +
+ ". The broker wants a version between " + feature.minSupportedVersion() + " and " +
feature.maxSupportedVersion() + ", inclusive.");
}
// A feature is not found in the finalizedFeature map if it is unknown to the controller or set to 0 (feature not enabled).
- // Only log if the feature name is not known by the controller.
- if (!Features.PRODUCTION_FEATURE_NAMES.contains(feature.name()))
- log.warn("Broker {} registered with feature {} that is unknown to the controller",
- brokerId, feature.name());
+ if (!finalizedFeatures.featureNames().contains(feature.name()))
+ log.debug("Broker {} registered with version range ({}, {}] of feature {} which controller does not know " +
+ "or has finalized version of 0.",
+ brokerId, feature.minSupportedVersion(), feature.maxSupportedVersion(), feature.name());
return new BrokerFeature().
setName(feature.name()).
setMinSupportedVersion(feature.minSupportedVersion()).
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 56cbb80d540..3f7bb4df74c 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -119,6 +119,7 @@ import org.apache.kafka.raft.RaftClient;
import org.apache.kafka.server.authorizer.AclCreateResult;
import org.apache.kafka.server.authorizer.AclDeleteResult;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.fault.FaultHandler;
import org.apache.kafka.server.fault.FaultHandlerException;
@@ -2171,11 +2172,14 @@ public final class QuorumController implements Controller {
ControllerRequestContext context,
BrokerRegistrationRequestData request
) {
+ // populate finalized features map with latest known kraft version for validation
+ Map<String, Short> controllerFeatures = new HashMap<>(featureControl.finalizedFeatures(Long.MAX_VALUE).featureMap());
+ controllerFeatures.put(KRaftVersion.FEATURE_NAME, raftClient.kraftVersion().featureLevel());
return appendWriteEvent("registerBroker", context.deadlineNs(),
() -> {
ControllerResult<BrokerRegistrationReply> result = clusterControl.
- registerBroker(request, offsetControl.nextWriteOffset(), featureControl.
- finalizedFeatures(Long.MAX_VALUE));
+ registerBroker(request, offsetControl.nextWriteOffset(),
+ new FinalizedControllerFeatures(controllerFeatures, Long.MAX_VALUE));
rescheduleMaybeFenceStaleBrokers();
return result;
},
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
index ddc7506b0d2..0646d4aaa56 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.BrokerRegistrationRequestData;
import org.apache.kafka.common.message.ControllerRegistrationRequestData;
import org.apache.kafka.common.metadata.BrokerRegistrationChangeRecord;
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
import org.apache.kafka.common.metadata.FenceBrokerRecord;
import org.apache.kafka.common.metadata.RegisterBrokerRecord;
import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerEndpoint;
@@ -49,7 +50,9 @@ import org.apache.kafka.metadata.placement.PartitionAssignment;
import org.apache.kafka.metadata.placement.PlacementSpec;
import org.apache.kafka.metadata.placement.UsableBroker;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.TestFeatureVersion;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.junit.jupiter.api.Test;
@@ -62,9 +65,11 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Stream;
@@ -531,6 +536,135 @@ public class ClusterControlManagerTest {
clusterControl.brokerRegistrations().get(2).toRecord(options));
}
+ @Test
+ public void testRegistrationWithUnsupportedFeature() {
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ Map<String, VersionRange> supportedFeatures = new HashMap<>();
+ supportedFeatures.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
+ MetadataVersion.IBP_3_1_IV0.featureLevel(),
+ MetadataVersion.IBP_3_7_IV0.featureLevel()));
+ supportedFeatures.put(TestFeatureVersion.FEATURE_NAME, VersionRange.of(
+ TestFeatureVersion.TEST_0.featureLevel(),
+ TestFeatureVersion.TEST_1.featureLevel()));
+ FeatureControlManager featureControl = new FeatureControlManager.Builder().
+ setSnapshotRegistry(snapshotRegistry).
+ setQuorumFeatures(new QuorumFeatures(0, supportedFeatures, Collections.singletonList(0))).
+ setMetadataVersion(MetadataVersion.IBP_3_7_IV0).
+ build();
+ ClusterControlManager clusterControl = new ClusterControlManager.Builder().
+ setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
+ setTime(new MockTime(0, 0, 0)).
+ setSnapshotRegistry(snapshotRegistry).
+ setFeatureControlManager(featureControl).
+ setBrokerUncleanShutdownHandler((brokerId, records) -> { }).
+ build();
+ clusterControl.activate();
+ FeatureLevelRecord testFeatureRecord = new FeatureLevelRecord().
+ setName(TestFeatureVersion.FEATURE_NAME).setFeatureLevel((short) 1);
+ featureControl.replay(testFeatureRecord);
+
+ List<Uuid> logDirs = asList(Uuid.fromString("yJGxmjfbQZSVFAlNM3uXZg"), Uuid.fromString("Mj3CW3OSRi29cFeNJlXuAQ"));
+ BrokerRegistrationRequestData baseRequest = new BrokerRegistrationRequestData().
+ setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
+ setBrokerId(0).
+ setRack(null).
+ setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")).
+ setLogDirs(logDirs);
+
+ assertEquals("Unable to register because the broker does not support finalized version 1 of " +
+ "test.feature.version. The broker wants a version between 0 and 0, inclusive.",
+ assertThrows(UnsupportedVersionException.class,
+ () -> clusterControl.registerBroker(
+ baseRequest.setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
+ Collections.singleton(new BrokerRegistrationRequestData.Feature().
+ setName(MetadataVersion.FEATURE_NAME).
+ setMinSupportedVersion(MetadataVersion.IBP_3_1_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_7_IV0.featureLevel())).iterator())),
+ 123L,
+ featureControl.finalizedFeatures(Long.MAX_VALUE))).getMessage());
+ }
+
+ @Test
+ public void testRegistrationWithUnsupportedKraftVersion() {
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ Map<String, VersionRange> supportedFeatures = new HashMap<>();
+ supportedFeatures.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
+ MetadataVersion.IBP_3_1_IV0.featureLevel(),
+ MetadataVersion.IBP_3_9_IV0.featureLevel()));
+ supportedFeatures.put(KRaftVersion.FEATURE_NAME, VersionRange.of(
+ KRaftVersion.KRAFT_VERSION_1.featureLevel(),
+ KRaftVersion.KRAFT_VERSION_1.featureLevel()));
+ FeatureControlManager featureControl = new FeatureControlManager.Builder().
+ setSnapshotRegistry(snapshotRegistry).
+ setQuorumFeatures(new QuorumFeatures(0, supportedFeatures, Collections.singletonList(0))).
+ setMetadataVersion(MetadataVersion.IBP_3_9_IV0).
+ build();
+ ClusterControlManager clusterControl = new ClusterControlManager.Builder().
+ setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
+ setTime(new MockTime(0, 0, 0)).
+ setSnapshotRegistry(snapshotRegistry).
+ setFeatureControlManager(featureControl).
+ setBrokerUncleanShutdownHandler((brokerId, records) -> { }).
+ build();
+ clusterControl.activate();
+
+ List<Uuid> logDirs = asList(Uuid.fromString("yJGxmjfbQZSVFAlNM3uXZg"), Uuid.fromString("Mj3CW3OSRi29cFeNJlXuAQ"));
+ BrokerRegistrationRequestData baseRequest = new BrokerRegistrationRequestData().
+ setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
+ setBrokerId(0).
+ setRack(null).
+ setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")).
+ setLogDirs(logDirs);
+
+ // quorum controller passes in the latest kraft version to populate finalized features
+ Map<String, Short> updatedFeaturesMap = new HashMap<>(featureControl.finalizedFeatures(Long.MAX_VALUE).featureMap());
+ updatedFeaturesMap.put(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel());
+ FinalizedControllerFeatures updatedFinalizedFeatures = new FinalizedControllerFeatures(updatedFeaturesMap, Long.MAX_VALUE);
+
+ assertEquals("Unable to register because the broker does not support finalized version 1 of " +
+ "kraft.version. The broker wants a version between 0 and 0, inclusive.",
+ assertThrows(UnsupportedVersionException.class,
+ () -> clusterControl.registerBroker(
+ baseRequest.setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
+ Collections.singleton(new BrokerRegistrationRequestData.Feature().
+ setName(MetadataVersion.FEATURE_NAME).
+ setMinSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel())).iterator())),
+ 123L,
+ updatedFinalizedFeatures)).getMessage());
+
+ assertEquals("Unable to register because the broker does not support finalized version 1 of " +
+ "kraft.version. The broker wants a version between 0 and 0, inclusive.",
+ assertThrows(UnsupportedVersionException.class,
+ () -> clusterControl.registerBroker(
+ baseRequest.setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
+ Arrays.asList(
+ new BrokerRegistrationRequestData.Feature().
+ setName(MetadataVersion.FEATURE_NAME).
+ setMinSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel()),
+ new BrokerRegistrationRequestData.Feature().
+ setName(KRaftVersion.FEATURE_NAME).
+ setMinSupportedVersion(KRaftVersion.KRAFT_VERSION_0.featureLevel()).
+ setMaxSupportedVersion(KRaftVersion.KRAFT_VERSION_0.featureLevel())).iterator())),
+ 123L,
+ updatedFinalizedFeatures)).getMessage());
+
+ clusterControl.registerBroker(
+ baseRequest.setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
+ Arrays.asList(
+ new BrokerRegistrationRequestData.Feature().
+ setName(MetadataVersion.FEATURE_NAME).
+ setMinSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_9_IV0.featureLevel()),
+ new BrokerRegistrationRequestData.Feature().
+ setName(KRaftVersion.FEATURE_NAME).
+ setMinSupportedVersion(KRaftVersion.KRAFT_VERSION_1.featureLevel()).
+ setMaxSupportedVersion(KRaftVersion.KRAFT_VERSION_1.featureLevel())).iterator())),
+ 123L,
+ updatedFinalizedFeatures);
+ }
+
@Test
public void testRegistrationWithUnsupportedMetadataVersion() {
SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
@@ -552,8 +686,8 @@ public class ClusterControlManagerTest {
build();
clusterControl.activate();
- assertEquals("Unable to register because the broker does not support version 4 of " +
- "metadata.version. It wants a version between 1 and 1, inclusive.",
+ assertEquals("Unable to register because the broker does not support finalized version 4 of " +
+ "metadata.version. The broker wants a version between 1 and 1, inclusive.",
assertThrows(UnsupportedVersionException.class,
() -> clusterControl.registerBroker(
new BrokerRegistrationRequestData().
@@ -564,8 +698,8 @@ public class ClusterControlManagerTest {
123L,
featureControl.finalizedFeatures(Long.MAX_VALUE))).getMessage());
- assertEquals("Unable to register because the broker does not support version 4 of " +
- "metadata.version. It wants a version between 7 and 7, inclusive.",
+ assertEquals("Unable to register because the broker does not support finalized version 4 of " +
+ "metadata.version. The broker wants a version between 7 and 7, inclusive.",
assertThrows(UnsupportedVersionException.class,
() -> clusterControl.registerBroker(
new BrokerRegistrationRequestData().
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 23f8bdbd66b..08213dfdfee 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.BrokerIdNotRegisteredException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.AllocateProducerIdsRequestData;
import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData;
import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData.ReassignableTopic;
@@ -106,6 +107,8 @@ import org.apache.kafka.metalog.LocalLogManagerTestEnv;
import org.apache.kafka.raft.Batch;
import org.apache.kafka.raft.OffsetAndEpoch;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.Features;
+import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.common.TopicIdPartition;
import org.apache.kafka.server.fault.FaultHandlerException;
@@ -117,6 +120,7 @@ import org.apache.kafka.timeline.SnapshotRegistry;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
@@ -685,6 +689,61 @@ public class QuorumControllerTest {
}
}
+ @ParameterizedTest
+ @CsvSource(value = {"0, 0", "0, 1", "1, 0", "1, 1"})
+ public void testRegisterBrokerKRaftVersions(short finalizedKraftVersion, short brokerMaxSupportedKraftVersion) throws Throwable {
+ try (
+ LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).
+ setLastKRaftVersion(KRaftVersion.fromFeatureLevel(finalizedKraftVersion)).
+ build();
+ QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).
+ setControllerBuilderInitializer(controllerBuilder ->
+ controllerBuilder.setConfigSchema(SCHEMA)).
+ setBootstrapMetadata(SIMPLE_BOOTSTRAP).
+ build()
+ ) {
+ ListenerCollection listeners = new ListenerCollection();
+ listeners.add(new Listener().setName("PLAINTEXT").
+ setHost("localhost").setPort(9092));
+ QuorumController active = controlEnv.activeController();
+ BrokerRegistrationRequestData.FeatureCollection brokerFeatures = new BrokerRegistrationRequestData.FeatureCollection();
+ brokerFeatures.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(MetadataVersion.IBP_3_0_IV1.featureLevel())
+ .setMaxSupportedVersion(MetadataVersion.latestTesting().featureLevel()));
+ // broker registration requests do not include initial versions of features
+ if (brokerMaxSupportedKraftVersion != 0) {
+ brokerFeatures.add(new BrokerRegistrationRequestData.Feature()
+ .setName(KRaftVersion.FEATURE_NAME)
+ .setMinSupportedVersion(Features.KRAFT_VERSION.minimumProduction())
+ .setMaxSupportedVersion(brokerMaxSupportedKraftVersion));
+ }
+ BrokerRegistrationRequestData request = new BrokerRegistrationRequestData().
+ setBrokerId(0).
+ setClusterId(active.clusterId()).
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
+ setFeatures(brokerFeatures).
+ setLogDirs(Collections.singletonList(Uuid.fromString("vBpaRsZVSaGsQT53wtYGtg"))).
+ setListeners(listeners);
+
+ if (brokerMaxSupportedKraftVersion < finalizedKraftVersion) {
+ Throwable exception = assertThrows(ExecutionException.class, () -> active.registerBroker(
+ ANONYMOUS_CONTEXT,
+ request).get());
+ assertEquals(UnsupportedVersionException.class, exception.getCause().getClass());
+ assertEquals("Unable to register because the broker does not support finalized version " +
+ finalizedKraftVersion + " of kraft.version. The broker wants a version between 0 and " +
+ brokerMaxSupportedKraftVersion + ", inclusive.",
+ exception.getCause().getMessage());
+ } else {
+ BrokerRegistrationReply reply = active.registerBroker(
+ ANONYMOUS_CONTEXT,
+ request).get();
+ assertTrue(reply.epoch() >= 5, "Unexpected broker epoch " + reply.epoch());
+ }
+ }
+ }
+
@Test
public void testUnregisterBroker() throws Throwable {
try (
diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
index 2f24d3f966b..add04f06030 100644
--- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
+++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
@@ -470,6 +470,11 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
*/
private final EventQueue eventQueue;
+ /**
+ * The latest kraft version used by this local log manager.
+ */
+ private final KRaftVersion lastKRaftVersion;
+
/**
* Whether this LocalLogManager has been shut down.
*/
@@ -499,13 +504,15 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
public LocalLogManager(LogContext logContext,
int nodeId,
SharedLogData shared,
- String threadNamePrefix) {
+ String threadNamePrefix,
+ KRaftVersion lastKRaftVersion) {
this.log = logContext.logger(LocalLogManager.class);
this.nodeId = nodeId;
this.shared = shared;
this.maxReadOffset = shared.initialMaxReadOffset();
this.eventQueue = new KafkaEventQueue(Time.SYSTEM, logContext,
threadNamePrefix, new ShutdownEvent());
+ this.lastKRaftVersion = lastKRaftVersion;
shared.registerLogManager(this);
}
@@ -840,6 +847,6 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
@Override
public KRaftVersion kraftVersion() {
- return KRaftVersion.KRAFT_VERSION_0;
+ return lastKRaftVersion;
}
}
diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java
index dd6ba56a3b1..661c96a3a8b 100644
--- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java
+++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java
@@ -25,6 +25,7 @@ import org.apache.kafka.metalog.LocalLogManager.LocalRecordBatch;
import org.apache.kafka.metalog.LocalLogManager.SharedLogData;
import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.snapshot.RawSnapshotReader;
import org.apache.kafka.test.TestUtils;
@@ -71,6 +72,7 @@ public class LocalLogManagerTestEnv implements AutoCloseable {
private final int numManagers;
private Optional<RawSnapshotReader> snapshotReader = Optional.empty();
private Consumer<SharedLogData> sharedLogDataInitializer = __ -> { };
+ private KRaftVersion lastKRaftVersion = KRaftVersion.KRAFT_VERSION_0;
public Builder(int numManagers) {
this.numManagers = numManagers;
@@ -86,11 +88,20 @@ public class LocalLogManagerTestEnv implements AutoCloseable {
return this;
}
+ /**
+ * Used to mock the latest KRaft version that would be returned from RaftClient.kraftVersion()
+ */
+ public Builder setLastKRaftVersion(KRaftVersion kraftVersion) {
+ this.lastKRaftVersion = kraftVersion;
+ return this;
+ }
+
public LocalLogManagerTestEnv build() {
return new LocalLogManagerTestEnv(
numManagers,
snapshotReader,
- sharedLogDataInitializer);
+ sharedLogDataInitializer,
+ lastKRaftVersion);
}
public LocalLogManagerTestEnv buildWithMockListeners() {
@@ -114,7 +125,8 @@ public class LocalLogManagerTestEnv implements AutoCloseable {
private LocalLogManagerTestEnv(
int numManagers,
Optional<RawSnapshotReader> snapshotReader,
- Consumer<SharedLogData> sharedLogDataInitializer
+ Consumer<SharedLogData> sharedLogDataInitializer,
+ KRaftVersion lastKRaftVersion
) {
clusterId = Uuid.randomUuid().toString();
dir = TestUtils.tempDirectory();
@@ -127,7 +139,8 @@ public class LocalLogManagerTestEnv implements AutoCloseable {
new LogContext(String.format("[LocalLogManager %d] ", nodeId)),
nodeId,
shared,
- String.format("LocalLogManager-%d_", nodeId)));
+ String.format("LocalLogManager-%d_", nodeId),
+ lastKRaftVersion));
}
} catch (Throwable t) {
for (LocalLogManager logManager : newLogManagers) {