kowshik commented on a change in pull request #8680:
URL: https://github.com/apache/kafka/pull/8680#discussion_r436493426
##########
File path: core/src/main/scala/kafka/cluster/Broker.scala
##########
@@ -34,14 +36,21 @@ object Broker {
brokerId: Int,
endpoints: util.List[Endpoint],
interBrokerEndpoint: Endpoint)
extends AuthorizerServerInfo
+
+ def apply(id: Int, endPoints: Seq[EndPoint], rack: Option[String]): Broker =
{
+ new Broker(id, endPoints, rack, emptySupportedFeatures)
+ }
}
/**
* A Kafka broker.
- * A broker has an id, a collection of end-points, an optional rack and a
listener to security protocol map.
- * Each end-point is (host, port, listenerName).
+ *
+ * @param id a broker id
+ * @param endPoints a collection of EndPoint. Each end-point is (host, port,
listener name, security protocol).
+ * @param rack an optional rack
+ * @param features optional supported features
Review comment:
Done.
##########
File path: core/src/main/scala/kafka/server/KafkaServer.scala
##########
@@ -660,6 +674,10 @@ class KafkaServer(val config: KafkaConfig, time: Time =
Time.SYSTEM, threadNameP
if (zkClient != null)
CoreUtils.swallow(zkClient.close(), this)
+ if (featureChangeListener != null) {
+ CoreUtils.swallow(featureChangeListener.close(), this)
Review comment:
Done.
##########
File path: core/src/test/scala/kafka/zk/FeatureZNodeTest.scala
##########
@@ -0,0 +1,111 @@
+package kafka.zk
+
+import java.nio.charset.StandardCharsets
+
+import org.apache.kafka.common.feature.{Features, FinalizedVersionRange}
+import org.apache.kafka.common.feature.Features._
+import org.junit.Assert.{assertEquals, assertThrows}
+import org.junit.Test
+
+import scala.jdk.CollectionConverters._
+
+class FeatureZNodeTest {
+
+ @Test
+ def testEncodeDecode(): Unit = {
+ val featureZNode = FeatureZNode(
+ FeatureZNodeStatus.Enabled,
+ Features.finalizedFeatures(
+ Map[String, FinalizedVersionRange](
+ "feature1" -> new FinalizedVersionRange(1, 2),
+ "feature2" -> new FinalizedVersionRange(2, 4)).asJava))
+ val decoded = FeatureZNode.decode(FeatureZNode.encode(featureZNode))
+ assertEquals(featureZNode.status, decoded.status)
+ assertEquals(featureZNode.features, decoded.features)
Review comment:
Done.
##########
File path:
core/src/test/scala/unit/kafka/server/FinalizedFeatureChangeListenerTest.scala
##########
@@ -0,0 +1,185 @@
+package kafka.server
+
+import kafka.zk.{FeatureZNode, FeatureZNodeStatus, ZkVersion,
ZooKeeperTestHarness}
+import kafka.utils.{Exit, TestUtils}
+import org.apache.kafka.common.feature.{Features, FinalizedVersionRange,
SupportedVersionRange}
+import org.apache.kafka.common.internals.FatalExitError
+import org.junit.Assert.{assertEquals, assertFalse, assertNotEquals,
assertThrows, assertTrue}
+import org.junit.{Before, Test}
+
+import scala.concurrent.TimeoutException
+import scala.jdk.CollectionConverters._
+
+class FinalizedFeatureChangeListenerTest extends ZooKeeperTestHarness {
+ @Before
+ override def setUp(): Unit = {
+ super.setUp()
+ FinalizedFeatureCache.clear()
+ SupportedFeatures.clear()
+ }
+
+ private def createSupportedFeatures(): Features[SupportedVersionRange] = {
+ val supportedFeaturesMap = Map[String, SupportedVersionRange](
+ "feature_1" -> new SupportedVersionRange(1, 4),
+ "feature_2" -> new SupportedVersionRange(1, 3))
+
SupportedFeatures.update(Features.supportedFeatures(supportedFeaturesMap.asJava))
+ SupportedFeatures.get
+ }
+
+ private def createFinalizedFeatures(): FinalizedFeaturesAndEpoch = {
+ val finalizedFeaturesMap = Map[String, FinalizedVersionRange](
+ "feature_1" -> new FinalizedVersionRange(2, 3))
+ val finalizedFeatures =
Features.finalizedFeatures(finalizedFeaturesMap.asJava)
+ zkClient.createFeatureZNode(FeatureZNode(FeatureZNodeStatus.Enabled,
finalizedFeatures))
+ val (mayBeFeatureZNodeBytes, version) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(version, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeBytes.isEmpty)
+ FinalizedFeaturesAndEpoch(finalizedFeatures, version)
+ }
+
+ private def createListener(expectedCacheContent:
Option[FinalizedFeaturesAndEpoch]): FinalizedFeatureChangeListener = {
+ val listener = new FinalizedFeatureChangeListener(zkClient)
+ assertFalse(listener.isListenerInitiated)
+ assertTrue(FinalizedFeatureCache.isEmpty)
+ listener.initOrThrow(15000)
+ assertTrue(listener.isListenerInitiated)
+ if (expectedCacheContent.isDefined) {
+ val mayBeNewCacheContent = FinalizedFeatureCache.get
+ assertFalse(mayBeNewCacheContent.isEmpty)
+ val newCacheContent = mayBeNewCacheContent.get
+ assertEquals(expectedCacheContent.get.features, newCacheContent.features)
+ assertEquals(expectedCacheContent.get.epoch, newCacheContent.epoch)
+ } else {
+ val mayBeNewCacheContent = FinalizedFeatureCache.get
+ assertTrue(mayBeNewCacheContent.isEmpty)
+ }
+ listener
+ }
+
+ /**
+ * Tests that the listener can be initialized, and that it can listen to ZK
notifications
+ * successfully from an "Enabled" FeatureZNode (the ZK data has no feature
incompatibilities).
+ */
+ @Test
+ def testInitSuccessAndNotificationSuccess(): Unit = {
+ createSupportedFeatures()
+ val initialFinalizedFeatures = createFinalizedFeatures()
+ val listener = createListener(Some(initialFinalizedFeatures))
+
+ val updatedFinalizedFeaturesMap = Map[String, FinalizedVersionRange](
+ "feature_1" -> new FinalizedVersionRange(2, 4))
+ val updatedFinalizedFeatures =
Features.finalizedFeatures(updatedFinalizedFeaturesMap.asJava)
+ zkClient.updateFeatureZNode(FeatureZNode(FeatureZNodeStatus.Enabled,
updatedFinalizedFeatures))
+ val (mayBeFeatureZNodeNewBytes, updatedVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(updatedVersion, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeNewBytes.isEmpty)
+ assertTrue(updatedVersion > initialFinalizedFeatures.epoch)
+ TestUtils.waitUntilTrue(() => {
+
FinalizedFeatureCache.get.get.equals(FinalizedFeaturesAndEpoch(updatedFinalizedFeatures,
updatedVersion))
+ }, "Timed out waiting for FinalizedFeatureCache to be updated with new
features")
+ assertTrue(listener.isListenerInitiated)
+ }
+
+ /**
+ * Tests that the listener can be initialized, and that it can process
FeatureZNode deletion
+ * successfully.
+ */
+ @Test
+ def testFeatureZNodeDeleteNotificationProcessing(): Unit = {
+ createSupportedFeatures()
+ val initialFinalizedFeatures = createFinalizedFeatures()
+ val listener = createListener(Some(initialFinalizedFeatures))
+
+ zkClient.deleteFeatureZNode()
+ val (mayBeFeatureZNodeDeletedBytes, deletedVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertEquals(deletedVersion, ZkVersion.UnknownVersion)
+ assertTrue(mayBeFeatureZNodeDeletedBytes.isEmpty)
+ TestUtils.waitUntilTrue(() => {
+ FinalizedFeatureCache.isEmpty
+ }, "Timed out waiting for FinalizedFeatureCache to become empty")
+ assertTrue(listener.isListenerInitiated)
+ }
+
+ /**
+ * Tests that the listener can be initialized, and that it can process
disabling of a FeatureZNode
+ * successfully.
+ */
+ @Test
+ def testFeatureZNodeDisablingNotificationProcessing(): Unit = {
+ createSupportedFeatures()
+ val initialFinalizedFeatures = createFinalizedFeatures()
+ val listener = createListener(Some(initialFinalizedFeatures))
+
+ val updatedFinalizedFeaturesMap = Map[String, FinalizedVersionRange]()
+ val updatedFinalizedFeatures =
Features.finalizedFeatures(updatedFinalizedFeaturesMap.asJava)
+ zkClient.updateFeatureZNode(FeatureZNode(FeatureZNodeStatus.Disabled,
updatedFinalizedFeatures))
+ val (mayBeFeatureZNodeNewBytes, updatedVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(updatedVersion, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeNewBytes.isEmpty)
+ assertTrue(updatedVersion > initialFinalizedFeatures.epoch)
+ TestUtils.waitUntilTrue(() => {
+ FinalizedFeatureCache.get.isDefined &&
+
FinalizedFeatureCache.get.get.features.equals(updatedFinalizedFeatures) &&
Review comment:
Done. I have modified the code such that
`FeatureCacheUpdater.updateLatestOrThrow` will now clear the cache whenever it
sees that the feature ZK node is disabled.
Great point!
##########
File path: core/src/main/scala/kafka/server/FinalizedFeatureChangeListener.scala
##########
@@ -0,0 +1,231 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import java.util.concurrent.{CountDownLatch, LinkedBlockingQueue, TimeUnit}
+
+import kafka.utils.{Logging, ShutdownableThread}
+import kafka.zk.{FeatureZNode,FeatureZNodeStatus, KafkaZkClient, ZkVersion}
+import kafka.zookeeper.ZNodeChangeHandler
+import org.apache.kafka.common.internals.FatalExitError
+
+import scala.concurrent.TimeoutException
+import scala.util.control.Exception.ignoring
+
+/**
+ * Listens to changes in the ZK feature node, via the ZK client. Whenever a
change notification
+ * is received from ZK, the feature cache in FinalizedFeatureCache is
asynchronously updated
+ * to the latest features read from ZK. The cache updates are serialized
through a single
+ * notification processor thread.
+ *
+ * @param zkClient the Zookeeper client
+ */
+class FinalizedFeatureChangeListener(zkClient: KafkaZkClient) extends Logging {
+
+ /**
+ * Helper class used to update the FinalizedFeatureCache.
+ *
+ * @param featureZkNodePath the path to the ZK feature node to be read
+ * @param maybeNotifyOnce an optional latch that can be used to notify
the caller when an
+ * updateOrThrow() operation is over
+ */
+ private class FeatureCacheUpdater(featureZkNodePath: String,
maybeNotifyOnce: Option[CountDownLatch]) {
+
+ def this(featureZkNodePath: String) = this(featureZkNodePath, Option.empty)
+
+ /**
+ * Updates the feature cache in FinalizedFeatureCache with the latest
features read from the
+ * ZK node in featureZkNodePath. If the cache update is not successful,
then, a suitable
+ * exception is raised.
+ *
+ * NOTE: if a notifier was provided in the constructor, then, this method
can be invoked exactly
+ * once successfully. A subsequent invocation will raise an exception.
+ *
+ * @throws IllegalStateException, if a non-empty notifier was provided
in the constructor, and
+ * this method is called again after a successful previous
invocation.
+ * @throws FeatureCacheUpdateException, if there was an error in
updating the
+ * FinalizedFeatureCache.
+ * @throws RuntimeException, if there was a failure in
reading/deserializing the
+ * contents of the feature ZK node.
+ */
+ def updateLatestOrThrow(): Unit = {
+ maybeNotifyOnce.foreach(notifier => {
+ if (notifier.getCount != 1) {
+ throw new IllegalStateException(
+ "Can not notify after updateLatestOrThrow was called more than
once successfully.")
+ }
+ })
+
+ debug(s"Reading feature ZK node at path: $featureZkNodePath")
+ val (mayBeFeatureZNodeBytes, version) =
zkClient.getDataAndVersion(featureZkNodePath)
+
+ // There are 4 cases:
+ //
+ // (empty dataBytes, valid version) => The empty dataBytes will
fail FeatureZNode deserialization.
+ // FeatureZNode, when present
in ZK, can not have empty contents.
+ // (non-empty dataBytes, valid version) => This is a valid case, and
should pass FeatureZNode deserialization
+ // if dataBytes contains valid
data.
+ // (empty dataBytes, unknown version) => This is a valid case, and
this can happen if the FeatureZNode
+ // does not exist in ZK.
+ // (non-empty dataBytes, unknown version) => This case is impossible,
since, KafkaZkClient.getDataAndVersion
+ // API ensures that unknown
version is returned only when the
+ // ZK node is absent.
Therefore dataBytes should be empty in such
+ // a case.
+ if (version == ZkVersion.UnknownVersion) {
+ info(s"Feature ZK node at path: $featureZkNodePath does not exist")
+ FinalizedFeatureCache.clear()
+ } else {
+ val featureZNode = FeatureZNode.decode(mayBeFeatureZNodeBytes.get)
+ featureZNode.status match {
+ case FeatureZNodeStatus.Disabled => {
+ info(s"Feature ZK node at path: $featureZkNodePath is in disabled
status.")
+ FinalizedFeatureCache.updateOrThrow(featureZNode.features, version)
+ }
+ case FeatureZNodeStatus.Enabled => {
+ FinalizedFeatureCache.updateOrThrow(featureZNode.features, version)
+ }
+ case _ => throw new IllegalStateException(s"Unexpected
FeatureZNodeStatus found in $featureZNode")
+ }
+ }
+
+ maybeNotifyOnce.foreach(notifier => notifier.countDown())
+ }
+
+ /**
+ * Waits until at least a single updateLatestOrThrow completes
successfully. This method returns
+ * immediately if an updateLatestOrThrow call had already completed
successfully.
+ *
+ * @param waitTimeMs the timeout for the wait operation
+ *
+ * @throws TimeoutException if the wait can not be completed
in waitTimeMs
+ * milli seconds
+ */
+ def awaitUpdateOrThrow(waitTimeMs: Long): Unit = {
+ maybeNotifyOnce.foreach(notifier => {
+ if (!notifier.await(waitTimeMs, TimeUnit.MILLISECONDS)) {
+ throw new TimeoutException(
+ s"Timed out after waiting for ${waitTimeMs}ms for FeatureCache to
be updated.")
+ }
+ })
+ }
+ }
+
+ /**
+ * A shutdownable thread to process feature node change notifications that
are populated into the
+ * queue. If any change notification can not be processed successfully
(unless it is due to an
+ * interrupt), the thread treats it as a fatal event and triggers Broker
exit.
+ *
+ * @param name name of the thread
+ */
+ private class ChangeNotificationProcessorThread(name: String) extends
ShutdownableThread(name = name) {
+ override def doWork(): Unit = {
+ try {
+ ignoring(classOf[InterruptedException]) {
+ queue.take.updateLatestOrThrow()
+ }
+ } catch {
+ case e: Exception => {
+ error("Failed to process feature ZK node change event. The broker
will eventually exit.", e)
+ throw new FatalExitError(1)
+ }
+ }
+ }
+ }
+
+ // Feature ZK node change handler.
+ object FeatureZNodeChangeHandler extends ZNodeChangeHandler {
+ override val path: String = FeatureZNode.path
+
+ override def handleCreation(): Unit = {
+ info(s"Feature ZK node created at path: $path")
+ queue.add(new FeatureCacheUpdater(path))
+ }
+
+ override def handleDataChange(): Unit = {
+ info(s"Feature ZK node updated at path: $path")
+ queue.add(new FeatureCacheUpdater(path))
+ }
+
+ override def handleDeletion(): Unit = {
+ warn(s"Feature ZK node deleted at path: $path")
+ // This event may happen, rarely (ex: ZK corruption or operational
error).
+ // In such a case, we prefer to just log a warning and treat the case as
if the node is absent,
+ // and populate the FinalizedFeatureCache with empty finalized features.
+ queue.add(new FeatureCacheUpdater(path))
+ }
+ }
+
+ private val queue = new LinkedBlockingQueue[FeatureCacheUpdater]
+
+ private val thread = new
ChangeNotificationProcessorThread("feature-zk-node-event-process-thread")
+
+ /**
+ * This method initializes the feature ZK node change listener. Optionally,
it also ensures to
+ * update the FinalizedFeatureCache once with the latest contents of the
feature ZK node
+ * (if the node exists). This step helps ensure that feature
incompatibilities (if any) in brokers
+ * are conveniently detected before the initOrThrow() method returns to the
caller. If feature
+ * incompatibilities are detected, this method will throw an Exception to
the caller, and the Broker
+ * will exit eventually.
+ *
+ * @param waitOnceForCacheUpdateMs # of milli seconds to wait for feature
cache to be updated once.
+ * If this parameter <= 0, no wait
operation happens.
+ *
+ * @throws Exception if feature incompatibility check could not be finished
in a timely manner
+ */
+ def initOrThrow(waitOnceForCacheUpdateMs: Long): Unit = {
+ if (waitOnceForCacheUpdateMs <= 0) {
+ throw new IllegalArgumentException(
+ s"Expected waitOnceForCacheUpdateMs > 0, but provided:
$waitOnceForCacheUpdateMs")
+ }
+
+ thread.start()
Review comment:
Done.
##########
File path:
core/src/test/scala/unit/kafka/server/FinalizedFeatureChangeListenerTest.scala
##########
@@ -0,0 +1,185 @@
+package kafka.server
Review comment:
Done.
##########
File path: core/src/test/scala/unit/kafka/server/SupportedFeaturesTest.scala
##########
@@ -0,0 +1,39 @@
+package kafka.server
Review comment:
Done.
##########
File path: core/src/test/scala/unit/kafka/server/FinalizedFeatureCacheTest.scala
##########
@@ -0,0 +1,99 @@
+package kafka.server
Review comment:
Done.
##########
File path: core/src/test/scala/kafka/zk/FeatureZNodeTest.scala
##########
@@ -0,0 +1,111 @@
+package kafka.zk
Review comment:
Done.
##########
File path:
core/src/test/scala/unit/kafka/server/FinalizedFeatureChangeListenerTest.scala
##########
@@ -0,0 +1,185 @@
+package kafka.server
+
+import kafka.zk.{FeatureZNode, FeatureZNodeStatus, ZkVersion,
ZooKeeperTestHarness}
+import kafka.utils.{Exit, TestUtils}
+import org.apache.kafka.common.feature.{Features, FinalizedVersionRange,
SupportedVersionRange}
+import org.apache.kafka.common.internals.FatalExitError
+import org.junit.Assert.{assertEquals, assertFalse, assertNotEquals,
assertThrows, assertTrue}
+import org.junit.{Before, Test}
+
+import scala.concurrent.TimeoutException
+import scala.jdk.CollectionConverters._
+
+class FinalizedFeatureChangeListenerTest extends ZooKeeperTestHarness {
+ @Before
+ override def setUp(): Unit = {
+ super.setUp()
+ FinalizedFeatureCache.clear()
+ SupportedFeatures.clear()
+ }
+
+ private def createSupportedFeatures(): Features[SupportedVersionRange] = {
+ val supportedFeaturesMap = Map[String, SupportedVersionRange](
+ "feature_1" -> new SupportedVersionRange(1, 4),
+ "feature_2" -> new SupportedVersionRange(1, 3))
+
SupportedFeatures.update(Features.supportedFeatures(supportedFeaturesMap.asJava))
+ SupportedFeatures.get
+ }
+
+ private def createFinalizedFeatures(): FinalizedFeaturesAndEpoch = {
+ val finalizedFeaturesMap = Map[String, FinalizedVersionRange](
+ "feature_1" -> new FinalizedVersionRange(2, 3))
+ val finalizedFeatures =
Features.finalizedFeatures(finalizedFeaturesMap.asJava)
+ zkClient.createFeatureZNode(FeatureZNode(FeatureZNodeStatus.Enabled,
finalizedFeatures))
+ val (mayBeFeatureZNodeBytes, version) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(version, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeBytes.isEmpty)
+ FinalizedFeaturesAndEpoch(finalizedFeatures, version)
+ }
+
+ private def createListener(expectedCacheContent:
Option[FinalizedFeaturesAndEpoch]): FinalizedFeatureChangeListener = {
+ val listener = new FinalizedFeatureChangeListener(zkClient)
+ assertFalse(listener.isListenerInitiated)
+ assertTrue(FinalizedFeatureCache.isEmpty)
+ listener.initOrThrow(15000)
+ assertTrue(listener.isListenerInitiated)
+ if (expectedCacheContent.isDefined) {
+ val mayBeNewCacheContent = FinalizedFeatureCache.get
+ assertFalse(mayBeNewCacheContent.isEmpty)
+ val newCacheContent = mayBeNewCacheContent.get
+ assertEquals(expectedCacheContent.get.features, newCacheContent.features)
+ assertEquals(expectedCacheContent.get.epoch, newCacheContent.epoch)
+ } else {
+ val mayBeNewCacheContent = FinalizedFeatureCache.get
+ assertTrue(mayBeNewCacheContent.isEmpty)
+ }
+ listener
+ }
+
+ /**
+ * Tests that the listener can be initialized, and that it can listen to ZK
notifications
+ * successfully from an "Enabled" FeatureZNode (the ZK data has no feature
incompatibilities).
+ */
+ @Test
+ def testInitSuccessAndNotificationSuccess(): Unit = {
+ createSupportedFeatures()
+ val initialFinalizedFeatures = createFinalizedFeatures()
+ val listener = createListener(Some(initialFinalizedFeatures))
+
+ val updatedFinalizedFeaturesMap = Map[String, FinalizedVersionRange](
+ "feature_1" -> new FinalizedVersionRange(2, 4))
+ val updatedFinalizedFeatures =
Features.finalizedFeatures(updatedFinalizedFeaturesMap.asJava)
+ zkClient.updateFeatureZNode(FeatureZNode(FeatureZNodeStatus.Enabled,
updatedFinalizedFeatures))
+ val (mayBeFeatureZNodeNewBytes, updatedVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(updatedVersion, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeNewBytes.isEmpty)
+ assertTrue(updatedVersion > initialFinalizedFeatures.epoch)
+ TestUtils.waitUntilTrue(() => {
+
FinalizedFeatureCache.get.get.equals(FinalizedFeaturesAndEpoch(updatedFinalizedFeatures,
updatedVersion))
+ }, "Timed out waiting for FinalizedFeatureCache to be updated with new
features")
+ assertTrue(listener.isListenerInitiated)
+ }
+
+ /**
+ * Tests that the listener can be initialized, and that it can process
FeatureZNode deletion
+ * successfully.
+ */
+ @Test
+ def testFeatureZNodeDeleteNotificationProcessing(): Unit = {
+ createSupportedFeatures()
+ val initialFinalizedFeatures = createFinalizedFeatures()
+ val listener = createListener(Some(initialFinalizedFeatures))
+
+ zkClient.deleteFeatureZNode()
+ val (mayBeFeatureZNodeDeletedBytes, deletedVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertEquals(deletedVersion, ZkVersion.UnknownVersion)
+ assertTrue(mayBeFeatureZNodeDeletedBytes.isEmpty)
+ TestUtils.waitUntilTrue(() => {
+ FinalizedFeatureCache.isEmpty
+ }, "Timed out waiting for FinalizedFeatureCache to become empty")
+ assertTrue(listener.isListenerInitiated)
+ }
+
+ /**
+ * Tests that the listener can be initialized, and that it can process
disabling of a FeatureZNode
+ * successfully.
+ */
+ @Test
+ def testFeatureZNodeDisablingNotificationProcessing(): Unit = {
+ createSupportedFeatures()
+ val initialFinalizedFeatures = createFinalizedFeatures()
+ val listener = createListener(Some(initialFinalizedFeatures))
+
+ val updatedFinalizedFeaturesMap = Map[String, FinalizedVersionRange]()
+ val updatedFinalizedFeatures =
Features.finalizedFeatures(updatedFinalizedFeaturesMap.asJava)
+ zkClient.updateFeatureZNode(FeatureZNode(FeatureZNodeStatus.Disabled,
updatedFinalizedFeatures))
+ val (mayBeFeatureZNodeNewBytes, updatedVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(updatedVersion, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeNewBytes.isEmpty)
+ assertTrue(updatedVersion > initialFinalizedFeatures.epoch)
+ TestUtils.waitUntilTrue(() => {
+ FinalizedFeatureCache.get.isDefined &&
+
FinalizedFeatureCache.get.get.features.equals(updatedFinalizedFeatures) &&
+ FinalizedFeatureCache.get.get.epoch == updatedVersion
+ }, "Timed out waiting for FinalizedFeatureCache to become empty")
+ assertTrue(listener.isListenerInitiated)
+ }
+
+ /**
+ * Tests that the listener initialization fails when it picks up a feature
incompatibility from
+ * ZK from an "Enabled" FeatureZNode.
+ */
+ @Test
+ def testInitFailureDueToFeatureIncompatibility(): Unit = {
+ createSupportedFeatures()
+
+ val incompatibleFinalizedFeaturesMap = Map[String, FinalizedVersionRange](
+ "feature_1" -> new FinalizedVersionRange(2, 5))
+ val incompatibleFinalizedFeatures =
Features.finalizedFeatures(incompatibleFinalizedFeaturesMap.asJava)
+ zkClient.createFeatureZNode(FeatureZNode(FeatureZNodeStatus.Enabled,
incompatibleFinalizedFeatures))
+ val (mayBeFeatureZNodeBytes, initialVersion) =
zkClient.getDataAndVersion(FeatureZNode.path)
+ assertNotEquals(initialVersion, ZkVersion.UnknownVersion)
+ assertFalse(mayBeFeatureZNodeBytes.isEmpty)
+
+ Exit.setExitProcedure((status, _) => throw new FatalExitError(status))
Review comment:
Done. Changed it to use a latch that gets notified when the exit
procedure is called.
Great point!
##########
File path: core/src/main/scala/kafka/zk/ZkData.scala
##########
@@ -744,6 +781,165 @@ object DelegationTokenInfoZNode {
def decode(bytes: Array[Byte]): Option[TokenInformation] =
DelegationTokenManager.fromBytes(bytes)
}
+/**
+ * Represents the status of the FeatureZNode.
+ *
+ * Enabled -> This status means the feature versioning system (KIP-584) is
enabled, and, the
+ * finalized features stored in the FeatureZNode are active. This
status is written by
+ * the controller to the FeatureZNode only when the broker IBP
config is greater than
+ * or equal to KAFKA_2_6_IV1.
+ *
+ * Disabled -> This status means the feature versioning system (KIP-584) is
disabled, and, the
+ * the finalized features stored in the FeatureZNode is not
relevant. This status is
+ * written by the controller to the FeatureZNode only when the
broker IBP config
+ * is less than KAFKA_2_6_IV1.
+ *
+ * The purpose behind the FeatureZNodeStatus is that it helps differentiates
between the following
+ * cases:
+ *
+ * 1. New cluster bootstrap:
+ * For a new Kafka cluster (i.e. it is deployed first time), we would like
to start the cluster
+ * with all the possible supported features finalized immediately. The new
cluster will almost
+ * never be started with an old IBP config that’s less than KAFKA_2_6_IV1.
In such a case, the
+ * controller will start up and notice that the FeatureZNode is absent in
the new cluster.
+ * To handle the requirement, the controller will create a FeatureZNode
(with enabled status)
+ * containing the entire list of supported features as its finalized
features.
+ *
+ * 2. Cluster upgrade:
+ * Imagine there is an existing Kafka cluster with IBP config less than
KAFKA_2_6_IV1, but
+ * the Broker binary has been upgraded to a state where it supports the
feature versioning
+ * system (KIP-584). This means the user is upgrading from an earlier
version of the Broker
+ * binary. In this case, we want to start with no finalized features and
allow the user to enable
+ * them whenever they are ready i.e. in the future whenever the user sets
IBP config
+ * to be greater than or equal to KAFKA_2_6_IV1. The reason is that
enabling all the possible
+ * features immediately after an upgrade could be harmful to the cluster.
+ * In such a case:
+ * - Before the Broker upgrade (i.e. IBP config set to less than
KAFKA_2_6_IV1), the controller
+ * will start up and check if the FeatureZNode is absent. If true, then
it will react by
+ * creating a FeatureZNode with disabled status and empty features.
+ * - After the Broker upgrade (i.e. IBP config set to greater than or
equal to KAFKA_2_6_IV1),
+ * when the controller starts up it will check if the FeatureZNode
exists and whether it is
+ * disabled. In such a case, it won’t upgrade all features immediately.
Instead it will just
+ * switch the FeatureZNode status to enabled status. This lets the user
finalize the features
+ * later.
+ *
+ * 2. Cluster downgrade:
+ * Imagine that a Kafka cluster exists already and the IBP config is
greater than or equal to
+ * KAFKA_2_6_IV1. Then, the user decided to downgrade the cluster by
setting IBP config to a
+ * value less than KAFKA_2_6_IV1. This means the user is also disabling the
feature versioning
+ * system (KIP-584). In this case, when the controller starts up with the
lower IBP config, it
+ * will switch the FeatureZNode status to disabled with empty features.
+ */
+object FeatureZNodeStatus extends Enumeration {
+ val Disabled, Enabled = Value
+
+ def withNameOpt(value: Int): Option[Value] = {
+ values.find(_.id == value)
+ }
+}
+
+/**
+ * Represents the contents of the ZK node containing finalized feature
information.
+ *
+ * @param status the status of the ZK node
+ * @param features the cluster-wide finalized features
+ */
+case class FeatureZNode(status: FeatureZNodeStatus.Value, features:
Features[FinalizedVersionRange]) {
+}
+
+object FeatureZNode {
+ private val VersionKey = "version"
+ private val StatusKey = "status"
+ private val FeaturesKey = "features"
+
+ // V0 contains 'version', 'status' and 'features' keys.
+ val V0 = 0
+ val CurrentVersion = V0
+
+ def path = "/feature"
+
+ def asJavaMap(scalaMap: Map[String, Map[String, Short]]): util.Map[String,
util.Map[String, java.lang.Short]] = {
+ scalaMap
+ .view.mapValues(_.view.mapValues(scalaShort =>
java.lang.Short.valueOf(scalaShort)).toMap.asJava)
+ .toMap
+ .asJava
+ }
+
+ /**
+ * Encodes a FeatureZNode to JSON.
+ *
+ * @param featureZNode FeatureZNode to be encoded
+ *
+ * @return JSON representation of the FeatureZNode, as an
Array[Byte]
+ */
+ def encode(featureZNode: FeatureZNode): Array[Byte] = {
+ val jsonMap = collection.mutable.Map(
+ VersionKey -> CurrentVersion,
+ StatusKey -> featureZNode.status.id,
+ FeaturesKey -> featureZNode.features.toMap)
+ Json.encodeAsBytes(jsonMap.asJava)
+ }
+
+ /**
+ * Decodes the contents of the feature ZK node from Array[Byte] to a
FeatureZNode.
+ *
+ * @param jsonBytes the contents of the feature ZK node
+ *
+ * @return the FeatureZNode created from jsonBytes
+ *
+ * @throws IllegalArgumentException if the Array[Byte] can not be decoded.
+ */
+ def decode(jsonBytes: Array[Byte]): FeatureZNode = {
+ Json.tryParseBytes(jsonBytes) match {
+ case Right(js) =>
+ val featureInfo = js.asJsonObject
+ val version = featureInfo(VersionKey).to[Int]
+ if (version < V0 || version > CurrentVersion) {
Review comment:
Done. Great point!
##########
File path:
clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
##########
@@ -135,7 +179,42 @@ public static ApiVersionsResponse
createApiVersionsResponse(int throttleTimeMs,
data.setThrottleTimeMs(throttleTimeMs);
data.setErrorCode(Errors.NONE.code());
data.setApiKeys(apiKeys);
+
data.setSupportedFeatures(createSupportedFeatureKeys(latestSupportedFeatures));
+ if (finalizedFeatures.isPresent()) {
+
data.setFinalizedFeatures(createFinalizedFeatureKeys(finalizedFeatures.get()));
+ }
+ if (finalizedFeaturesEpoch.isPresent()) {
+ data.setFinalizedFeaturesEpoch(finalizedFeaturesEpoch.get());
Review comment:
Done.
##########
File path:
clients/src/test/java/org/apache/kafka/common/feature/FeaturesTest.java
##########
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.feature;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+
+public class FeaturesTest {
+
+ @Test
+ public void testEmptyFeatures() {
+ Map<String, Map<String, Short>> emptyMap = new HashMap<>();
+
+ Features<FinalizedVersionRange> emptyFinalizedFeatures =
Features.emptyFinalizedFeatures();
+ assertTrue(emptyFinalizedFeatures.features().isEmpty());
+ assertTrue(emptyFinalizedFeatures.toMap().isEmpty());
+ assertEquals(emptyFinalizedFeatures,
Features.fromFinalizedFeaturesMap(emptyMap));
+
+ Features<SupportedVersionRange> emptySupportedFeatures =
Features.emptySupportedFeatures();
+ assertTrue(emptySupportedFeatures.features().isEmpty());
+ assertTrue(emptySupportedFeatures.toMap().isEmpty());
+ assertEquals(emptySupportedFeatures,
Features.fromSupportedFeaturesMap(emptyMap));
+ }
+
+ @Test
+ public void testNullFeatures() {
+ assertThrows(
+ NullPointerException.class,
+ () -> Features.finalizedFeatures(null));
+ assertThrows(
+ NullPointerException.class,
+ () -> Features.supportedFeatures(null));
+ }
+
+ @Test
+ public void testGetAllFeaturesAPI() {
+ SupportedVersionRange v1 = new SupportedVersionRange((short) 1,
(short) 2);
+ SupportedVersionRange v2 = new SupportedVersionRange((short) 3,
(short) 4);
+ Map<String, SupportedVersionRange> allFeatures =
+ mkMap(mkEntry("feature_1", v1), mkEntry("feature_2", v2));
+ Features<SupportedVersionRange> features =
Features.supportedFeatures(allFeatures);
+ assertEquals(allFeatures, features.features());
+ }
+
+ @Test
+ public void testGetAPI() {
+ SupportedVersionRange v1 = new SupportedVersionRange((short) 1,
(short) 2);
+ SupportedVersionRange v2 = new SupportedVersionRange((short) 3,
(short) 4);
+ Map<String, SupportedVersionRange> allFeatures =
mkMap(mkEntry("feature_1", v1), mkEntry("feature_2", v2));
+ Features<SupportedVersionRange> features =
Features.supportedFeatures(allFeatures);
+ assertEquals(v1, features.get("feature_1"));
+ assertEquals(v2, features.get("feature_2"));
+ assertNull(features.get("nonexistent_feature"));
+ }
+
+ @Test
+ public void testFromFeaturesMapToFeaturesMap() {
+ SupportedVersionRange v1 = new SupportedVersionRange((short) 1,
(short) 2);
+ SupportedVersionRange v2 = new SupportedVersionRange((short) 3,
(short) 4);
+ Map<String, SupportedVersionRange> allFeatures =
mkMap(mkEntry("feature_1", v1), mkEntry("feature_2", v2));
+
+ Features<SupportedVersionRange> features =
Features.supportedFeatures(allFeatures);
+
+ Map<String, Map<String, Short>> expected = mkMap(
+ mkEntry("feature_1", mkMap(mkEntry("min_version", (short) 1),
mkEntry("max_version", (short) 2))),
+ mkEntry("feature_2", mkMap(mkEntry("min_version", (short) 3),
mkEntry("max_version", (short) 4))));
+ assertEquals(expected, features.toMap());
+ assertEquals(features, Features.fromSupportedFeaturesMap(expected));
+ }
+
+ @Test
+ public void testFromToFinalizedFeaturesMap() {
+ FinalizedVersionRange v1 = new FinalizedVersionRange((short) 1,
(short) 2);
+ FinalizedVersionRange v2 = new FinalizedVersionRange((short) 3,
(short) 4);
+ Map<String, FinalizedVersionRange> allFeatures =
mkMap(mkEntry("feature_1", v1), mkEntry("feature_2", v2));
+
+ Features<FinalizedVersionRange> features =
Features.finalizedFeatures(allFeatures);
+
+ Map<String, Map<String, Short>> expected = mkMap(
+ mkEntry("feature_1", mkMap(mkEntry("min_version_level", (short)
1), mkEntry("max_version_level", (short) 2))),
+ mkEntry("feature_2", mkMap(mkEntry("min_version_level", (short)
3), mkEntry("max_version_level", (short) 4))));
+ assertEquals(expected, features.toMap());
+ assertEquals(features, Features.fromFinalizedFeaturesMap(expected));
+ }
+
+ @Test
+ public void testToStringFinalizedFeatures() {
+ FinalizedVersionRange v1 = new FinalizedVersionRange((short) 1,
(short) 2);
+ FinalizedVersionRange v2 = new FinalizedVersionRange((short) 3,
(short) 4);
+ Map<String, FinalizedVersionRange> allFeatures =
mkMap(mkEntry("feature_1", v1), mkEntry("feature_2", v2));
+
+ Features<FinalizedVersionRange> features =
Features.finalizedFeatures(allFeatures);
+
+ assertEquals(
+ "Features{(feature_1 -> FinalizedVersionRange[min_version_level:1,
max_version_level:2]), (feature_2 -> FinalizedVersionRange[min_version_level:3,
max_version_level:4])}",
+ features.toString());
+ }
+
+ @Test
+ public void testToStringSupportedFeatures() {
+ SupportedVersionRange v1 = new SupportedVersionRange((short) 1,
(short) 2);
+ SupportedVersionRange v2 = new SupportedVersionRange((short) 3,
(short) 4);
+ Map<String, SupportedVersionRange> allFeatures
+ = mkMap(mkEntry("feature_1", v1), mkEntry("feature_2", v2));
+
+ Features<SupportedVersionRange> features =
Features.supportedFeatures(allFeatures);
+
+ assertEquals(
+ "Features{(feature_1 -> SupportedVersionRange[min_version:1,
max_version:2]), (feature_2 -> SupportedVersionRange[min_version:3,
max_version:4])}",
+ features.toString());
+ }
+
+ @Test
+ public void testFromToSupportedFeaturesMap() {
Review comment:
Done.
##########
File path:
clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
##########
@@ -113,14 +127,44 @@ public static ApiVersionsResponse fromStruct(Struct
struct, short version) {
}
}
- public static ApiVersionsResponse apiVersionsResponse(int throttleTimeMs,
byte maxMagic) {
+ public static ApiVersionsResponse apiVersionsResponse(
+ int throttleTimeMs,
+ byte maxMagic,
+ Features<SupportedVersionRange> latestSupportedFeatures) {
+ return apiVersionsResponse(
+ throttleTimeMs, maxMagic, latestSupportedFeatures,
Optional.empty(), Optional.empty());
+ }
+
+ public static ApiVersionsResponse apiVersionsResponse(
+ int throttleTimeMs,
+ byte maxMagic,
+ Features<SupportedVersionRange> latestSupportedFeatures,
+ Features<FinalizedVersionRange> finalizedFeatures,
+ int finalizedFeaturesEpoch) {
+ return apiVersionsResponse(
+ throttleTimeMs, maxMagic, latestSupportedFeatures,
Optional.of(finalizedFeatures), Optional.of(finalizedFeaturesEpoch));
+ }
+
+ private static ApiVersionsResponse apiVersionsResponse(
+ int throttleTimeMs,
+ byte maxMagic,
+ Features<SupportedVersionRange> latestSupportedFeatures,
+ Optional<Features<FinalizedVersionRange>> finalizedFeatures,
+ Optional<Integer> finalizedFeaturesEpoch) {
Review comment:
Done. I'm no longer passing 2 optionals, since, we decided (below) that
epoch can be set as -1 whenever it is absent.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
