KAFKA-18435 Remove zookeeper dependencies in build.gradle (#18450)

Remove Apache ZooKeeper from the Apache Kafka build. Also remove commons IO, commons CLI, and netty, which were dependencies we took only because of ZooKeeper.

In order to keep the size of this PR manageable, I did not remove all classes which formerly interfaced with ZK. I just removed the ZK types. Fortunately, Kafka generally wrapped ZK data structures rather than using them directly.

Some classes were pretty entangled with ZK, so it was easier just to stub them out. For ZkNodeChangeNotificationListener.scala, PartitionStateMachine.scala, ReplicaStateMachine.scala, KafkaZkClient.scala, and ZookeeperClient.scala, I replaced all the functions with "throw new UnsupportedOperationException". Since the tests for these classes have been removed, as well as the ZK-based broker code, this should be OK as an incremental step.

Reviewers: Chia-Ping Tsai <[email protected]>

Author: cmccabe

LICENSE-binary

@@ -208,10 +208,8 @@ License Version 2.0:
 audience-annotations-0.12.0
 caffeine-3.1.1
 commons-beanutils-1.9.4
-commons-cli-1.4
 commons-collections-3.2.2
 commons-digester-2.1
-commons-io-2.14.0
 commons-lang3-3.12.0
 commons-logging-1.3.2
 commons-validator-1.9.0
@@ -257,15 +255,6 @@ lz4-java-1.8.0
 maven-artifact-3.9.6
 metrics-core-4.1.12.1
 metrics-core-2.2.0
-netty-buffer-4.1.115.Final
-netty-codec-4.1.115.Final
-netty-common-4.1.115.Final
-netty-handler-4.1.115.Final
-netty-resolver-4.1.115.Final
-netty-transport-4.1.115.Final
-netty-transport-classes-epoll-4.1.115.Final
-netty-transport-native-epoll-4.1.115.Final
-netty-transport-native-unix-common-4.1.115.Final
 opentelemetry-proto-1.0.0-alpha
 plexus-utils-3.5.1
 rocksdbjni-7.9.2
@@ -275,8 +264,6 @@ scala-reflect-2.13.15
 snappy-java-1.1.10.5
 snakeyaml-2.2
 swagger-annotations-2.2.25
-zookeeper-3.8.4
-zookeeper-jute-3.8.4
 
 ===============================================================================
 This product bundles various third-party components under other open source

NOTICE-binary

@@ -596,45 +596,6 @@ limitations under the License.
 This software includes projects with other licenses -- see `doc/LICENSE.md`.
 
 
-Apache ZooKeeper - Server
-Copyright 2008-2021 The Apache Software Foundation
-
-This product includes software developed at
-The Apache Software Foundation (http://www.apache.org/).
-
-
-Apache ZooKeeper - Jute
-Copyright 2008-2021 The Apache Software Foundation
-
-This product includes software developed at
-The Apache Software Foundation (http://www.apache.org/).
-
-
-The Netty Project
-                            =================
-
-Please visit the Netty web site for more information:
-
-  * https://netty.io/
-
-Copyright 2014 The Netty Project
-
-The Netty Project 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:
-
-  https://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.
-
-Also, please refer to each LICENSE.<component>.txt file, which is located in
-the 'license' directory of the distribution file, for the license terms of the
-components that this product depends on.
-
 -------------------------------------------------------------------------------
 This product contains the extensions to Java Collections Framework which has
 been derived from the works by JSR-166 EG, Doug Lea, and Jason T. Greene:

build.gradle

@@ -190,10 +190,6 @@ allprojects {
           libs.scalaReflect,
           libs.jacksonAnnotations,
           libs.jacksonDatabindYaml,
-          // be explicit about the Netty dependency version instead of relying on the version set by
-          // ZooKeeper (potentially older and containing CVEs)
-          libs.nettyHandler,
-          libs.nettyTransportNativeEpoll,
           libs.log4j2Api,
           libs.log4j2Core,
           libs.log4j1Bridge2Api
@@ -1107,20 +1103,6 @@ project(':core') {
     // only needed transitively, but set it explicitly to ensure it has the same version as scala-library
     implementation libs.scalaReflect
     implementation libs.scalaLogging
-    implementation libs.commonsIo // ZooKeeper dependency. Do not use, this is going away.
-    implementation(libs.zookeeper) {
-      // Dropwizard Metrics are required by ZooKeeper as of v3.6.0,
-      // but the library should *not* be used in Kafka code
-      implementation libs.dropwizardMetrics
-      exclude module: 'slf4j-log4j12'
-      exclude module: 'log4j'
-      // Both Kafka and Zookeeper use slf4j. ZooKeeper moved from log4j to logback in v3.8.0.
-      // We are removing Zookeeper's dependency on logback so we have a singular logging backend.
-      exclude module: 'logback-classic'
-      exclude module: 'logback-core'
-    }
-    // ZooKeeperMain depends on commons-cli but declares the dependency as `provided`
-    implementation libs.commonsCli
     implementation log4jLibs
 
     runtimeOnly log4jRuntimeLibs

checkstyle/import-control.xml

@@ -451,7 +451,6 @@
       <subpackage name="internals">
         <allow pkg="com.fasterxml.jackson" />
         <allow pkg="kafka.utils" />
-        <allow pkg="org.apache.zookeeper" />
         <allow pkg="org.apache.log4j" />
       </subpackage>
     </subpackage>

core/src/main/scala/kafka/common/ZkNodeChangeNotificationListener.scala

@@ -16,18 +16,9 @@
  */
 package kafka.common
 
-import java.nio.charset.StandardCharsets.UTF_8
-import java.util.concurrent.LinkedBlockingQueue
-import java.util.concurrent.atomic.AtomicBoolean
-
 import kafka.utils.Logging
-import kafka.zk.{KafkaZkClient, StateChangeHandlers}
-import kafka.zookeeper.{StateChangeHandler, ZNodeChildChangeHandler}
+import kafka.zk.KafkaZkClient
 import org.apache.kafka.common.utils.Time
-import org.apache.kafka.server.util.ShutdownableThread
-
-import scala.collection.Seq
-import scala.util.{Failure, Try}
 
 /**
  * Handle the notificationMessage.
@@ -56,105 +47,20 @@ class ZkNodeChangeNotificationListener(private val zkClient: KafkaZkClient,
                                        private val notificationHandler: NotificationHandler,
                                        private val changeExpirationMs: Long = 15 * 60 * 1000,
                                        private val time: Time = Time.SYSTEM) extends Logging {
-  private var lastExecutedChange = -1L
-  private val queue = new LinkedBlockingQueue[ChangeNotification]
-  private val thread = new ChangeEventProcessThread(s"$seqNodeRoot-event-process-thread")
-  private val isClosed = new AtomicBoolean(false)
-
   def init(): Unit = {
-    zkClient.registerStateChangeHandler(ZkStateChangeHandler)
-    zkClient.registerZNodeChildChangeHandler(ChangeNotificationHandler)
-    addChangeNotification()
-    thread.start()
+    throw new UnsupportedOperationException()
   }
 
   def close(): Unit = {
-    isClosed.set(true)
-    zkClient.unregisterStateChangeHandler(ZkStateChangeHandler.name)
-    zkClient.unregisterZNodeChildChangeHandler(ChangeNotificationHandler.path)
-    queue.clear()
-    thread.shutdown()
+    throw new UnsupportedOperationException()
   }
 
-  /**
-   * Process notifications
-   */
-  private def processNotifications(): Unit = {
-    try {
-      val notifications = zkClient.getChildren(seqNodeRoot).sorted
-      if (notifications.nonEmpty) {
-        info(s"Processing notification(s) to $seqNodeRoot")
-        val now = time.milliseconds
-        for (notification <- notifications) {
-          val changeId = changeNumber(notification)
-          if (changeId > lastExecutedChange) {
-            processNotification(notification)
-            lastExecutedChange = changeId
-          }
-        }
-        purgeObsoleteNotifications(now, notifications)
-      }
-    } catch {
-      case e: InterruptedException => if (!isClosed.get) error(s"Error while processing notification change for path = $seqNodeRoot", e)
-      case e: Exception => error(s"Error while processing notification change for path = $seqNodeRoot", e)
-    }
-  }
+  object ZkStateChangeHandler {
+    val name: String = null
 
-  private def processNotification(notification: String): Unit = {
-    val changeZnode = seqNodeRoot + "/" + notification
-    val (data, _) = zkClient.getDataAndStat(changeZnode)
-    data match {
-      case Some(d) => Try(notificationHandler.processNotification(d)) match {
-        case Failure(e) => error(s"error processing change notification ${new String(d, UTF_8)} from $changeZnode", e)
-        case _ =>
-      }
-      case None => warn(s"read null data from $changeZnode")
+    def afterInitializingSession(): Unit = {
+      throw new UnsupportedOperationException()
     }
   }
-
-  private def addChangeNotification(): Unit = {
-    if (!isClosed.get && queue.peek() == null)
-      queue.put(new ChangeNotification)
-  }
-
-  private class ChangeNotification {
-    def process(): Unit = processNotifications()
-  }
-
-  /**
-   * Purges expired notifications.
-   *
-   * @param now
-   * @param notifications
-   */
-  private def purgeObsoleteNotifications(now: Long, notifications: Seq[String]): Unit = {
-    for (notification <- notifications.sorted) {
-      val notificationNode = seqNodeRoot + "/" + notification
-      val (data, stat) = zkClient.getDataAndStat(notificationNode)
-      if (data.isDefined) {
-        if (now - stat.getCtime > changeExpirationMs) {
-          debug(s"Purging change notification $notificationNode")
-          zkClient.deletePath(notificationNode)
-        }
-      }
-    }
-  }
-
-  /* get the change number from a change notification znode */
-  private def changeNumber(name: String): Long = name.substring(seqNodePrefix.length).toLong
-
-  private class ChangeEventProcessThread(name: String) extends ShutdownableThread(name) {
-    override def doWork(): Unit = queue.take().process()
-  }
-
-  private object ChangeNotificationHandler extends ZNodeChildChangeHandler {
-    override val path: String = seqNodeRoot
-    override def handleChildChange(): Unit = addChangeNotification()
-  }
-
-  object ZkStateChangeHandler extends  StateChangeHandler {
-    override val name: String = StateChangeHandlers.zkNodeChangeListenerHandler(seqNodeRoot)
-    override def afterInitializingSession(): Unit = addChangeNotification()
-  }
 }
 

core/src/main/scala/kafka/controller/KafkaController.scala

@@ -28,10 +28,9 @@ import kafka.utils._
 import kafka.zk.KafkaZkClient.UpdateLeaderAndIsrResult
 import kafka.zk.TopicZNode.TopicIdReplicaAssignment
 import kafka.zk.{FeatureZNodeStatus, _}
-import kafka.zookeeper.{StateChangeHandler, ZNodeChangeHandler, ZNodeChildChangeHandler}
+import kafka.zookeeper.{StateChangeHandler, ZNodeChangeHandler, ZNodeChildChangeHandler, ZooKeeperClientException}
 import org.apache.kafka.clients.admin.FeatureUpdate.UpgradeType
 import org.apache.kafka.common.ElectionType
-import org.apache.kafka.common.KafkaException
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.errors.{BrokerNotAvailableException, ControllerMovedException, StaleBrokerEpochException}
@@ -45,8 +44,6 @@ import org.apache.kafka.server.BrokerFeatures
 import org.apache.kafka.server.common.{AdminOperationException, ProducerIdsBlock}
 import org.apache.kafka.server.metrics.KafkaMetricsGroup
 import org.apache.kafka.server.util.KafkaScheduler
-import org.apache.zookeeper.KeeperException
-import org.apache.zookeeper.KeeperException.Code
 
 import scala.collection.{Map, Seq, Set, immutable, mutable}
 import scala.collection.mutable.ArrayBuffer
@@ -1068,21 +1065,7 @@ class KafkaController(val config: KafkaConfig,
   }
 
   private def updateReplicaAssignmentForPartition(topicPartition: TopicPartition, assignment: ReplicaAssignment): Unit = {
-    val topicAssignment = mutable.Map() ++=
-      controllerContext.partitionFullReplicaAssignmentForTopic(topicPartition.topic) +=
-      (topicPartition -> assignment)
-
-    val setDataResponse = zkClient.setTopicAssignmentRaw(topicPartition.topic,
-      controllerContext.topicIds.get(topicPartition.topic),
-      topicAssignment, controllerContext.epochZkVersion)
-    setDataResponse.resultCode match {
-      case Code.OK =>
-        info(s"Successfully updated assignment of partition $topicPartition to $assignment")
-      case Code.NONODE =>
-        throw new IllegalStateException(s"Failed to update assignment for $topicPartition since the topic " +
-          "has no current assignment")
-      case _ => throw new KafkaException(setDataResponse.resultException.get)
-    }
+    throw new UnsupportedOperationException()
   }
 
   private def startNewReplicasForReassignedPartition(topicPartition: TopicPartition, newReplicas: Seq[Int]): Unit = {
@@ -1186,7 +1169,7 @@ class KafkaController(val config: KafkaConfig,
       try {
         zkClient.setOrCreatePartitionReassignment(updatedPartitionsBeingReassigned, controllerContext.epochZkVersion)
       } catch {
-        case e: KeeperException => throw new AdminOperationException(e)
+        case e: ZooKeeperClientException => throw new AdminOperationException(e)
       }
     }
   }