You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/05/16 09:46:42 UTC
(kafka) branch 3.7 updated: KAFKA-16539 Fix IncrementalAlterConfigs during ZK migration (#15744) (#15953)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 3.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push:
new b124aa48f29 KAFKA-16539 Fix IncrementalAlterConfigs during ZK migration (#15744) (#15953)
b124aa48f29 is described below
commit b124aa48f295106c5f9267f5c4cf81c7ba93fa18
Author: David Arthur <mu...@gmail.com>
AuthorDate: Thu May 16 05:46:35 2024 -0400
KAFKA-16539 Fix IncrementalAlterConfigs during ZK migration (#15744) (#15953)
This patch fixes two issues with IncrementalAlterConfigs and the ZK migration. First, it changes the handling of IncrementalAlterConfigs to check if the controller is ZK vs KRaft and only forward for KRaft. Second, it adds a check in KafkaZkClient#setOrCreateEntityConfigs to ensure a ZK broker is not directly modifying configs in ZK if there is a KRaft controller. This closes the race condition between KRaft taking over as the active controller and the ZK brokers learning about this.
*Forwarding*
During the ZK migration, there is a time when the ZK brokers are running with migrations enabled, but KRaft has yet to take over as the controller. Prior to KRaft taking over as the controller, the ZK brokers in migration mode were unconditionally forwarding IncrementalAlterConfigs (IAC) to the ZK controller. This works for some config types, but breaks when setting BROKER and BROKER_LOGGER configs for a specific broker. The behavior in KafkaApis for IAC was to always forward if the f [...]
With this fix, the IAC handler will check if the controller is KRaft or ZK and only forward for KRaft.
*Protected ZK Writes*
As part of KIP-500, we moved most (but not all) ZK mutations to the ZK controller. One of the things we did not move fully to the controller was entity configs. This is because there was some special logic that needed to run on the broker for certain config updates. If a broker-specific config was set, AdminClient would route the request to the proper broker. In KRaft, we have a different mechanism for handling broker-specific config updates.
Leaving this ZK update on the broker side would be okay if we were guarding writes on the controller epoch, but it turns out KafkaZkClient#setOrCreateEntityConfigs does unprotected "last writer wins" updates to ZK. This means a ZK broker could update the contents of ZK after the metadata had been migrated to KRaft. No good! To fix this, this patch adds a check on the controller epoch to KafkaZkClient#setOrCreateEntityConfigs but also adds logic to fail the update if the controller is [...]
The new logic in setOrCreateEntityConfigs adds STALE_CONTROLLER_EPOCH as a new exception that can be thrown while updating configs.
Reviewers: Luke Chen <sh...@gmail.com>, Akhilesh Chaganti <ak...@users.noreply.github.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../src/main/scala/kafka/admin/ConfigCommand.scala | 2 +-
.../scala/kafka/admin/ZkSecurityMigrator.scala | 2 +-
.../kafka/security/authorizer/AclAuthorizer.scala | 2 +-
core/src/main/scala/kafka/server/KafkaApis.scala | 12 ++-
core/src/main/scala/kafka/zk/KafkaZkClient.scala | 68 +++++++++++--
.../scala/integration/kafka/api/SaslSetup.scala | 2 +-
.../kafka/server/QuorumTestHarness.scala | 3 +-
.../kafka/zk/ZkMigrationIntegrationTest.scala | 110 ++++++++++++++++++---
.../scala/unit/kafka/zk/KafkaZkClientTest.scala | 24 ++++-
.../zk/migration/ZkConfigMigrationClientTest.scala | 1 +
.../kafka/jmh/server/PartitionCreationBench.java | 2 +-
11 files changed, 199 insertions(+), 29 deletions(-)
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index dd0151d4fb8..5e5f16c559e 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -114,7 +114,7 @@ object ConfigCommand extends Logging {
val zkClientConfig = ZkSecurityMigrator.createZkClientConfigFromOption(opts.options, opts.zkTlsConfigFile)
.getOrElse(new ZKClientConfig())
val zkClient = KafkaZkClient(zkConnectString, JaasUtils.isZkSaslEnabled || KafkaConfig.zkTlsClientAuthEnabled(zkClientConfig), 30000, 30000,
- Int.MaxValue, Time.SYSTEM, zkClientConfig = zkClientConfig, name = "ConfigCommand")
+ Int.MaxValue, Time.SYSTEM, zkClientConfig = zkClientConfig, name = "ConfigCommand", enableEntityConfigControllerCheck = false)
val adminZkClient = new AdminZkClient(zkClient)
try {
if (opts.options.has(opts.alterOpt))
diff --git a/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala b/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
index 7bc7339d548..07cbe45a8c7 100644
--- a/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
+++ b/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala
@@ -106,7 +106,7 @@ object ZkSecurityMigrator extends Logging {
val zkSessionTimeout = opts.options.valueOf(opts.zkSessionTimeoutOpt).intValue
val zkConnectionTimeout = opts.options.valueOf(opts.zkConnectionTimeoutOpt).intValue
val zkClient = KafkaZkClient(zkUrl, zkAcl, zkSessionTimeout, zkConnectionTimeout,
- Int.MaxValue, Time.SYSTEM, zkClientConfig = zkClientConfig, name = "ZkSecurityMigrator")
+ Int.MaxValue, Time.SYSTEM, zkClientConfig = zkClientConfig, name = "ZkSecurityMigrator", enableEntityConfigControllerCheck = false)
val enablePathCheck = opts.options.has(opts.enablePathCheckOpt)
val migrator = new ZkSecurityMigrator(zkClient)
migrator.run(enablePathCheck)
diff --git a/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala b/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala
index 34070d16378..5a8514020b4 100644
--- a/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala
+++ b/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala
@@ -208,7 +208,7 @@ class AclAuthorizer extends Authorizer with Logging {
// because such a cluster will not create any chroot path in ZooKeeper (it doesn't connect to ZooKeeper)
zkClient = KafkaZkClient(zkUrl, kafkaConfig.zkEnableSecureAcls, zkSessionTimeOutMs, zkConnectionTimeoutMs,
zkMaxInFlightRequests, time, name = "ACL authorizer", zkClientConfig = zkClientConfig,
- metricGroup = "kafka.security", metricType = "AclAuthorizer", createChrootIfNecessary = true)
+ metricGroup = "kafka.security", metricType = "AclAuthorizer", createChrootIfNecessary = true, enableEntityConfigControllerCheck = false)
zkClient.createAclPaths()
extendedAclSupport = kafkaConfig.interBrokerProtocolVersion.isAtLeast(IBP_2_0_IV1)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 872d5262c5c..1f90614ec3d 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -2970,6 +2970,15 @@ class KafkaApis(val requestChannel: RequestChannel,
(rType, rName) => authHelper.authorize(request.context, ALTER_CONFIGS, rType, rName))
val remaining = ConfigAdminManager.copyWithoutPreprocessed(original.data(), preprocessingResponses)
+ // Before deciding whether to forward or handle locally, a ZK broker needs to check if
+ // the active controller is ZK or KRaft. If the controller is KRaft, we need to forward.
+ // If the controller is ZK, we need to process the request locally.
+ val isKRaftController = metadataSupport match {
+ case ZkSupport(_, _, _, _, metadataCache, _) =>
+ metadataCache.getControllerId.exists(_.isInstanceOf[KRaftCachedControllerId])
+ case RaftSupport(_, _) => true
+ }
+
def sendResponse(secondPart: Option[ApiMessage]): Unit = {
secondPart match {
case Some(result: IncrementalAlterConfigsResponseData) =>
@@ -2982,9 +2991,10 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+ // Forwarding has not happened yet, so handle both ZK and KRaft cases here
if (remaining.resources().isEmpty) {
sendResponse(Some(new IncrementalAlterConfigsResponseData()))
- } else if ((!request.isForwarded) && metadataSupport.canForward()) {
+ } else if ((!request.isForwarded) && metadataSupport.canForward() && isKRaftController) {
metadataSupport.forwardingManager.get.forwardRequest(request,
new IncrementalAlterConfigsRequest(remaining, request.header.apiVersion()),
response => sendResponse(response.map(_.data())))
diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
index b15faffa5b7..65cf27ace8c 100644
--- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
+++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
@@ -59,8 +59,12 @@ case class SuccessfulRegistrationResult(zkControllerEpoch: Int, controllerEpochZ
* easier to migrate away from `ZkUtils` (since removed). We should revisit this. We should also consider whether a
* monolithic [[kafka.zk.ZkData]] is the way to go.
*/
-class KafkaZkClient private[zk] (zooKeeperClient: ZooKeeperClient, isSecure: Boolean, time: Time) extends AutoCloseable with
- Logging {
+class KafkaZkClient private[zk] (
+ zooKeeperClient: ZooKeeperClient,
+ isSecure: Boolean,
+ time: Time,
+ enableEntityConfigControllerCheck: Boolean
+) extends AutoCloseable with Logging {
private val metricsGroup: KafkaMetricsGroup = new KafkaMetricsGroup(this.getClass) {
override def metricName(name: String, metricTags: util.Map[String, String]): MetricName = {
@@ -458,6 +462,14 @@ class KafkaZkClient private[zk] (zooKeeperClient: ZooKeeperClient, isSecure: Boo
* Sets or creates the entity znode path with the given configs depending
* on whether it already exists or not.
*
+ * If enableEntityConfigControllerCheck is set, this method will ensure that a ZK controller is defined and
+ * that it is not modified within the duration of this call. This is done to prevent configs from being
+ * created or modified while the ZK to KRaft migration is taking place.
+ *
+ * The only case where enableEntityConfigControllerCheck should be false is when being called by ConfigCommand,
+ * i.e., "kafka-configs.sh --zookeeper". This is an old behavior we have kept around to allow users to setup
+ * SCRAM credentials and other configs before the cluster is started for the first time.
+ *
* If this is method is called concurrently, the last writer wins. In cases where we update configs and then
* partition assignment (i.e. create topic), it's possible for one thread to set this and the other to set the
* partition assignment. As such, the recommendation is to never call create topic for the same topic with different
@@ -466,13 +478,50 @@ class KafkaZkClient private[zk] (zooKeeperClient: ZooKeeperClient, isSecure: Boo
* @param rootEntityType entity type
* @param sanitizedEntityName entity name
* @throws KeeperException if there is an error while setting or creating the znode
- */
- def setOrCreateEntityConfigs(rootEntityType: String, sanitizedEntityName: String, config: Properties) = {
+ * @throws ControllerMovedException if no controller is defined, or a KRaft controller is defined
+ */
+ def setOrCreateEntityConfigs(rootEntityType: String, sanitizedEntityName: String, config: Properties): Unit = {
+ val controllerZkVersionOpt: Option[Int] = if (enableEntityConfigControllerCheck) {
+ val controllerRegistration = getControllerRegistration match {
+ case Some(registration) => registration
+ case None =>
+ // This case is mainly here to make tests less flaky (by virtue of retries).
+ // In practice, we always expect a /controller ZNode to exist
+ throw new ControllerMovedException(s"Cannot set entity configs for $rootEntityType $sanitizedEntityName " +
+ s"when there is no controller.")
+ }
+
+ // If there is a KRaft controller defined, don't even attempt this write. The broker will soon get a UMR
+ // from the new KRaft controller that lets it know about the new controller. It will then forward
+ // IncrementalAlterConfig requests instead of processing directly.
+ if (controllerRegistration.kraftEpoch.exists(epoch => epoch > 0)) {
+ throw new ControllerMovedException(s"Cannot set entity configs for $rootEntityType $sanitizedEntityName " +
+ s"directly when there is a KRaft controller.")
+ }
+
+ Some(controllerRegistration.zkVersion)
+ } else {
+ logger.warn("Setting entity configs without any checks on the controller.")
+ None
+ }
def set(configData: Array[Byte]): SetDataResponse = {
- val setDataRequest = SetDataRequest(ConfigEntityZNode.path(rootEntityType, sanitizedEntityName),
- configData, ZkVersion.MatchAnyVersion)
- retryRequestUntilConnected(setDataRequest)
+ // Since we're guarding against the controller switching to KRaft, we need to check that the controller hasn't
+ // changed during this method. We do that here by adding a CheckOp on the controller ZNode. The reason we
+ // don't use the controller epoch zkVersion here is that we can't consistently read the controller and
+ // controller epoch. This does _not_ guard against the existing "last writer wins" behavior of this method.
+ controllerZkVersionOpt match {
+ case Some(controllerZkVersion) =>
+ val multi = MultiRequest(Seq(
+ CheckOp(ControllerZNode.path, controllerZkVersion),
+ SetDataOp(ConfigEntityZNode.path(rootEntityType, sanitizedEntityName), configData, ZkVersion.MatchAnyVersion)
+ ))
+ val results = retryRequestUntilConnected(multi)
+ unwrapResponseWithControllerEpochCheck(results).asInstanceOf[SetDataResponse]
+ case None =>
+ val setDataRequest = SetDataRequest(ConfigEntityZNode.path(rootEntityType, sanitizedEntityName), configData, ZkVersion.MatchAnyVersion)
+ retryRequestUntilConnected(setDataRequest)
+ }
}
def createOrSet(configData: Array[Byte]): Unit = {
@@ -2229,7 +2278,8 @@ object KafkaZkClient {
zkClientConfig: ZKClientConfig,
metricGroup: String = "kafka.server",
metricType: String = "SessionExpireListener",
- createChrootIfNecessary: Boolean = false
+ createChrootIfNecessary: Boolean = false,
+ enableEntityConfigControllerCheck: Boolean = true
): KafkaZkClient = {
/* ZooKeeper 3.6.0 changed the default configuration for JUTE_MAXBUFFER from 4 MB to 1 MB.
@@ -2264,7 +2314,7 @@ object KafkaZkClient {
}
val zooKeeperClient = new ZooKeeperClient(connectString, sessionTimeoutMs, connectionTimeoutMs, maxInFlightRequests,
time, metricGroup, metricType, zkClientConfig, name)
- new KafkaZkClient(zooKeeperClient, isSecure, time)
+ new KafkaZkClient(zooKeeperClient, isSecure, time, enableEntityConfigControllerCheck)
}
// A helper function to transform a regular request into a MultiRequest
diff --git a/core/src/test/scala/integration/kafka/api/SaslSetup.scala b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
index 540c4eabb8d..1a8f555955a 100644
--- a/core/src/test/scala/integration/kafka/api/SaslSetup.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
@@ -202,7 +202,7 @@ trait SaslSetup {
val zkClientConfig = new ZKClientConfig()
TestUtils.resource(KafkaZkClient(
zkConnect, JaasUtils.isZkSaslEnabled || KafkaConfig.zkTlsClientAuthEnabled(zkClientConfig), 30000, 30000,
- Int.MaxValue, Time.SYSTEM, name = "SaslSetup", zkClientConfig = zkClientConfig)) { zkClient =>
+ Int.MaxValue, Time.SYSTEM, name = "SaslSetup", zkClientConfig = zkClientConfig, enableEntityConfigControllerCheck = false)) { zkClient =>
val adminZkClient = new AdminZkClient(zkClient)
val entityType = ConfigType.User
diff --git a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala
index 1a83809c0ad..893352a4b84 100755
--- a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala
+++ b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala
@@ -407,7 +407,8 @@ abstract class QuorumTestHarness extends Logging {
zkMaxInFlightRequests,
Time.SYSTEM,
name = "ZooKeeperTestHarness",
- new ZKClientConfig)
+ new ZKClientConfig,
+ enableEntityConfigControllerCheck = false)
adminZkClient = new AdminZkClient(zkClient)
} catch {
case t: Throwable =>
diff --git a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
index 75eab542031..c06fdc5e9dc 100644
--- a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
@@ -47,9 +47,10 @@ import org.apache.kafka.metadata.migration.ZkMigrationLeadershipState
import org.apache.kafka.raft.RaftConfig
import org.apache.kafka.server.ControllerRequestCompletionHandler
import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion, ProducerIdsBlock}
-import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotEquals, assertNotNull, assertTrue, fail}
+import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertEquals, assertFalse, assertNotEquals, assertNotNull, assertTrue, fail}
import org.junit.jupiter.api.{Assumptions, Timeout}
import org.junit.jupiter.api.extension.ExtendWith
+import org.junit.jupiter.api.function.Executable
import org.slf4j.{Logger, LoggerFactory}
import java.util
@@ -595,7 +596,12 @@ class ZkMigrationIntegrationTest {
val readyFuture = kraftCluster.controllers().values().asScala.head.controller.waitForReadyBrokers(3)
// Allocate a block of producer IDs while in ZK mode
- val nextProducerId = sendAllocateProducerIds(zkCluster.asInstanceOf[ZkClusterInstance]).get(30, TimeUnit.SECONDS)
+ var nextProducerId = -1L
+
+ TestUtils.retry(60000) {
+ assertDoesNotThrow((() => nextProducerId = sendAllocateProducerIds(zkCluster.asInstanceOf[ZkClusterInstance]).get(20, TimeUnit.SECONDS)): Executable)
+ }
+ assertEquals(0, nextProducerId)
// Enable migration configs and restart brokers
log.info("Restart brokers in migration mode")
@@ -617,16 +623,21 @@ class ZkMigrationIntegrationTest {
// Alter the metadata
log.info("Updating metadata with AdminClient")
admin = zkCluster.createAdminClient()
- alterTopicConfig(admin).all().get(60, TimeUnit.SECONDS)
- alterClientQuotas(admin).all().get(60, TimeUnit.SECONDS)
+ alterTopicConfig(admin)
+ alterClientQuotas(admin)
+ alterBrokerConfigs(admin)
// Verify the changes made to KRaft are seen in ZK
log.info("Verifying metadata changes with ZK")
verifyTopicConfigs(zkClient)
verifyClientQuotas(zkClient)
- val nextKRaftProducerId = sendAllocateProducerIds(zkCluster.asInstanceOf[ZkClusterInstance]).get(30, TimeUnit.SECONDS)
- assertNotEquals(nextProducerId, nextKRaftProducerId)
+ verifyBrokerConfigs(zkClient)
+ var nextKRaftProducerId = -1L
+ TestUtils.retry(60000) {
+ assertDoesNotThrow((() => nextKRaftProducerId = sendAllocateProducerIds(zkCluster.asInstanceOf[ZkClusterInstance]).get(20, TimeUnit.SECONDS)): Executable)
+ }
+ assertNotEquals(nextProducerId, nextKRaftProducerId)
} finally {
shutdownInSequence(zkCluster, kraftCluster)
}
@@ -683,7 +694,7 @@ class ZkMigrationIntegrationTest {
log.info("Updating metadata with AdminClient")
admin = zkCluster.createAdminClient()
alterUserScramCredentials(admin).all().get(60, TimeUnit.SECONDS)
- alterClientQuotas(admin).all().get(60, TimeUnit.SECONDS)
+ alterClientQuotas(admin)
// Verify the changes made to KRaft are seen in ZK
log.info("Verifying metadata changes with ZK")
@@ -849,6 +860,33 @@ class ZkMigrationIntegrationTest {
}
}
+ @ClusterTest(clusterType = Type.ZK, brokers = 3, metadataVersion = MetadataVersion.IBP_3_4_IV0, serverProperties = Array(
+ new ClusterConfigProperty(key = "inter.broker.listener.name", value = "EXTERNAL"),
+ new ClusterConfigProperty(key = "listeners", value = "PLAINTEXT://localhost:0,EXTERNAL://localhost:0"),
+ new ClusterConfigProperty(key = "advertised.listeners", value = "PLAINTEXT://localhost:0,EXTERNAL://localhost:0"),
+ new ClusterConfigProperty(key = "listener.security.protocol.map", value = "EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT"),
+ ))
+ def testIncrementalAlterConfigsPreMigration(zkCluster: ClusterInstance): Unit = {
+ // Enable migration configs and restart brokers without KRaft quorum ready
+ zkCluster.config().serverProperties().put(KafkaConfig.MigrationEnabledProp, "true")
+ zkCluster.config().serverProperties().put(KafkaConfig.QuorumVotersProp, "1@localhost:9999")
+ zkCluster.config().serverProperties().put(KafkaConfig.ControllerListenerNamesProp, "CONTROLLER")
+ zkCluster.config().serverProperties().put(KafkaConfig.ListenerSecurityProtocolMapProp, "CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT")
+ zkCluster.asInstanceOf[ZkClusterInstance].rollingBrokerRestart()
+ zkCluster.waitForReadyBrokers()
+
+ val admin = zkCluster.createAdminClient()
+ val zkClient = zkCluster.asInstanceOf[ZkClusterInstance].getUnderlying().zkClient
+ try {
+ alterBrokerConfigs(admin)
+ verifyBrokerConfigs(zkClient)
+ } finally {
+ admin.close()
+ zkClient.close()
+ zkCluster.stop()
+ }
+ }
+
def createTopic(topicName: String, numPartitions: Int, replicationFactor: Short, configs: util.Map[String, String], admin: Admin): Unit = {
val newTopic = new NewTopic(topicName, numPartitions, replicationFactor).configs(configs)
val createTopicResult = admin.createTopics(util.Collections.singletonList(newTopic))
@@ -948,16 +986,47 @@ class ZkMigrationIntegrationTest {
dataOpt.map(ProducerIdBlockZNode.parseProducerIdBlockData).get
}
- def alterTopicConfig(admin: Admin): AlterConfigsResult = {
+ def alterBrokerConfigs(admin: Admin): Unit = {
+ val defaultBrokerResource = new ConfigResource(ConfigResource.Type.BROKER, "")
+ val defaultBrokerConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(KafkaConfig.LogRetentionTimeMillisProp, "86400000"), AlterConfigOp.OpType.SET),
+ ).asJavaCollection
+ val broker0Resource = new ConfigResource(ConfigResource.Type.BROKER, "0")
+ val broker1Resource = new ConfigResource(ConfigResource.Type.BROKER, "1")
+ val specificBrokerConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(KafkaConfig.LogRetentionTimeMillisProp, "43200000"), AlterConfigOp.OpType.SET),
+ ).asJavaCollection
+
+ TestUtils.retry(60000) {
+ val result = admin.incrementalAlterConfigs(Map(
+ defaultBrokerResource -> defaultBrokerConfigs,
+ broker0Resource -> specificBrokerConfigs,
+ broker1Resource -> specificBrokerConfigs
+ ).asJava)
+ try {
+ result.all().get(10, TimeUnit.SECONDS)
+ } catch {
+ case t: Throwable => fail("Alter Broker Configs had an error", t)
+ }
+ }
+ }
+
+ def alterTopicConfig(admin: Admin): Unit = {
val topicResource = new ConfigResource(ConfigResource.Type.TOPIC, "test")
val alterConfigs = Seq(
new AlterConfigOp(new ConfigEntry(TopicConfig.SEGMENT_BYTES_CONFIG, "204800"), AlterConfigOp.OpType.SET),
new AlterConfigOp(new ConfigEntry(TopicConfig.SEGMENT_MS_CONFIG, null), AlterConfigOp.OpType.DELETE)
).asJavaCollection
- admin.incrementalAlterConfigs(Map(topicResource -> alterConfigs).asJava)
+ TestUtils.retry(60000) {
+ try {
+ admin.incrementalAlterConfigs(Map(topicResource -> alterConfigs).asJava).all().get(10, TimeUnit.SECONDS)
+ } catch {
+ case t: Throwable => fail("Alter Topic Configs had an error", t)
+ }
+ }
}
- def alterClientQuotas(admin: Admin): AlterClientQuotasResult = {
+ def alterClientQuotas(admin: Admin): Unit = {
val quotas = new util.ArrayList[ClientQuotaAlteration]()
quotas.add(new ClientQuotaAlteration(
new ClientQuotaEntity(Map("user" -> "user@1").asJava),
@@ -971,7 +1040,13 @@ class ZkMigrationIntegrationTest {
quotas.add(new ClientQuotaAlteration(
new ClientQuotaEntity(Map("ip" -> "8.8.8.8").asJava),
List(new ClientQuotaAlteration.Op("connection_creation_rate", 10.0)).asJava))
- admin.alterClientQuotas(quotas)
+ TestUtils.retry(60000) {
+ try {
+ admin.alterClientQuotas(quotas).all().get(10, TimeUnit.SECONDS)
+ } catch {
+ case t: Throwable => fail("Alter Client Quotas had an error", t)
+ }
+ }
}
def createUserScramCredentials(admin: Admin): AlterUserScramCredentialsResult = {
@@ -998,6 +1073,19 @@ class ZkMigrationIntegrationTest {
}
}
+ def verifyBrokerConfigs(zkClient: KafkaZkClient): Unit = {
+ TestUtils.retry(10000) {
+ val defaultBrokerProps = zkClient.getEntityConfigs(ConfigType.Broker, "<default>")
+ assertEquals("86400000", defaultBrokerProps.getProperty(KafkaConfig.LogRetentionTimeMillisProp))
+
+ val broker0Props = zkClient.getEntityConfigs(ConfigType.Broker, "0")
+ assertEquals("43200000", broker0Props.getProperty(KafkaConfig.LogRetentionTimeMillisProp))
+
+ val broker1Props = zkClient.getEntityConfigs(ConfigType.Broker, "1")
+ assertEquals("43200000", broker1Props.getProperty(KafkaConfig.LogRetentionTimeMillisProp))
+ }
+ }
+
def verifyClientQuotas(zkClient: KafkaZkClient): Unit = {
TestUtils.retry(10000) {
assertEquals("1000", zkClient.getEntityConfigs(ConfigType.User, Sanitizer.sanitize("user@1")).getProperty("consumer_byte_rate"))
diff --git a/core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala b/core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala
index 131b46d5671..1649d5df384 100644
--- a/core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala
@@ -82,7 +82,7 @@ class KafkaZkClientTest extends QuorumTestHarness {
zkClient.createControllerEpochRaw(1)
otherZkClient = KafkaZkClient(zkConnect, zkAclsEnabled.getOrElse(JaasUtils.isZkSaslEnabled), zkSessionTimeout,
zkConnectionTimeout, zkMaxInFlightRequests, Time.SYSTEM, name = "KafkaZkClient",
- zkClientConfig = new ZKClientConfig)
+ zkClientConfig = new ZKClientConfig, enableEntityConfigControllerCheck = false)
expiredSessionZkClient = ExpiredKafkaZkClient(zkConnect, zkAclsEnabled.getOrElse(JaasUtils.isZkSaslEnabled),
zkSessionTimeout, zkConnectionTimeout, zkMaxInFlightRequests, Time.SYSTEM)
}
@@ -1487,8 +1487,28 @@ class KafkaZkClientTest extends QuorumTestHarness {
}
}
+ @Test
+ def testNoEntityConfigUpdateWithNoController(): Unit = {
+ val props = new Properties()
+ props.put(KafkaConfig.DefaultReplicationFactorProp, "1") // normal config
+
+ // Create a client that cannot set entity configs without a controller present
+ val client = KafkaZkClient(zkConnect, zkAclsEnabled.getOrElse(JaasUtils.isZkSaslEnabled), zkSessionTimeout,
+ zkConnectionTimeout, zkMaxInFlightRequests, Time.SYSTEM, name = "KafkaZkClient",
+ zkClientConfig = new ZKClientConfig)
+
+ try {
+ assertThrows(
+ classOf[ControllerMovedException],
+ () => client.setOrCreateEntityConfigs(ConfigType.Broker, "1", props),
+ )
+ } finally {
+ client.close()
+ }
+ }
+
class ExpiredKafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean, time: Time)
- extends KafkaZkClient(zooKeeperClient, isSecure, time) {
+ extends KafkaZkClient(zooKeeperClient, isSecure, time, false) {
// Overwriting this method from the parent class to force the client to re-register the Broker.
override def shouldReCreateEphemeralZNode(ephemeralOwnerId: Long): Boolean = {
true
diff --git a/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala b/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala
index f04720399a8..b4bf3f6aa1e 100644
--- a/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala
@@ -250,6 +250,7 @@ class ZkConfigMigrationClientTest extends ZkMigrationTestHarness {
props.put(TopicConfig.RETENTION_MS_CONFIG, "300000")
zkClient.setOrCreateEntityConfigs(ConfigType.Topic, "test", props)
+ migrationState = migrationClient.claimControllerLeadership(migrationState)
migrationState = migrationClient.configClient().writeConfigs(new ConfigResource(ConfigResource.Type.TOPIC, "test"),
java.util.Collections.singletonMap(TopicConfig.SEGMENT_MS_CONFIG, "100000"), migrationState)
assertEquals(1, migrationState.migrationZkVersion())
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
index c40939235ff..86db11d4e36 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
@@ -144,7 +144,7 @@ public class PartitionCreationBench {
build();
scheduler.startup();
this.quotaManagers = QuotaFactory.instantiate(this.brokerProperties, this.metrics, this.time, "");
- this.zkClient = new KafkaZkClient(null, false, Time.SYSTEM) {
+ this.zkClient = new KafkaZkClient(null, false, Time.SYSTEM, false) {
@Override
public Properties getEntityConfigs(String rootEntityType, String sanitizedEntityName) {
return new Properties();