kowshik commented on a change in pull request #8680:
URL: https://github.com/apache/kafka/pull/8680#discussion_r437124061
##########
File path: core/src/main/scala/kafka/zk/ZkData.scala
##########
@@ -81,17 +83,26 @@ object BrokerIdsZNode {
object BrokerInfo {
/**
- * Create a broker info with v4 json format (which includes multiple
endpoints and rack) if
- * the apiVersion is 0.10.0.X or above. Register the broker with v2 json
format otherwise.
+ * - Create a broker info with v5 json format if the apiVersion is 2.6.x or
above.
+ * - Create a broker info with v4 json format (which includes multiple
endpoints and rack) if
+ * the apiVersion is 0.10.0.X or above but lesser than 2.6.x.
Review comment:
Done.
##########
File path: core/src/main/scala/kafka/zk/ZkData.scala
##########
@@ -81,17 +83,26 @@ object BrokerIdsZNode {
object BrokerInfo {
/**
- * Create a broker info with v4 json format (which includes multiple
endpoints and rack) if
- * the apiVersion is 0.10.0.X or above. Register the broker with v2 json
format otherwise.
+ * - Create a broker info with v5 json format if the apiVersion is 2.6.x or
above.
Review comment:
Done.
##########
File path: core/src/main/scala/kafka/server/FinalizedFeatureChangeListener.scala
##########
@@ -0,0 +1,243 @@
+/**
+ * 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.{StateChangeHandler, 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.clear()
+ }
+ 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))
+ }
+ }
+
+ object ZkStateChangeHandler extends StateChangeHandler {
+ val path: String = FeatureZNode.path
+
+ override val name: String = s"change-notification-$path"
Review comment:
Done.
----------------------------------------------------------------
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
