[LI-HOTFIX] Add topic deletion hook in zookeeper to trigger in-memory topic deletion flag changes:

TICKET =
LI_DESCRIPTION =

- Add TopicDeletionFlagPath `/topic_deletion_flag` to trigger changes in topic deletion enable flag in TopicDeletionManager.
- Controller will listen on this path when active and un-listen when resignation.
- If the znode is deleted, the topic deletion flag will be set according to the config.

RB=1306354
BUG=LIKAFKA-17224
G=Kafka-Code-Reviews
R=luwang
A=luwang

EXIT_CRITERIA = MANUAL ["After upstream makes delete.topic.enable a dynamic broker config"]

Author: hzxa21

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

@@ -98,7 +98,11 @@ object ControllerState {
     def value = 14
   }
 
+  case object TopicDeletionFlagChange extends ControllerState {
+    def value = 15
+  }
+
   val values: Seq[ControllerState] = Seq(Idle, ControllerChange, BrokerChange, TopicChange, TopicDeletion,
     PartitionReassignment, AutoLeaderBalance, ManualLeaderBalance, ControlledShutdown, IsrChange, LeaderAndIsrResponseReceived,
-    LogDirChange, ControllerShutdown, UncleanLeaderElectionEnable, TopicUncleanLeaderElectionEnable)
+    LogDirChange, ControllerShutdown, UncleanLeaderElectionEnable, TopicUncleanLeaderElectionEnable, TopicDeletionFlagChange)
 }

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

@@ -101,6 +101,7 @@ class KafkaController(val config: KafkaConfig,
   private val preferredReplicaElectionHandler = new PreferredReplicaElectionHandler(eventManager)
   private val isrChangeNotificationHandler = new IsrChangeNotificationHandler(eventManager)
   private val logDirEventNotificationHandler = new LogDirEventNotificationHandler(eventManager)
+  private val topicDeletionFlagHandler = new TopicDeletionFlagHandler(this, eventManager)
 
   @volatile private var activeControllerId = -1
   @volatile private var offlinePartitionCount = 0
@@ -244,7 +245,7 @@ class KafkaController(val config: KafkaConfig,
     val childChangeHandlers = Seq(brokerChangeHandler, topicChangeHandler, topicDeletionHandler, logDirEventNotificationHandler,
       isrChangeNotificationHandler)
     childChangeHandlers.foreach(zkClient.registerZNodeChildChangeHandler)
-    val nodeChangeHandlers = Seq(preferredReplicaElectionHandler, partitionReassignmentHandler)
+    val nodeChangeHandlers = Seq(preferredReplicaElectionHandler, partitionReassignmentHandler, topicDeletionFlagHandler)
     nodeChangeHandlers.foreach(zkClient.registerZNodeChangeHandlerAndCheckExistence)
 
     info("Deleting log dir event notifications")
@@ -325,6 +326,7 @@ class KafkaController(val config: KafkaConfig,
     zkClient.unregisterZNodeChildChangeHandler(topicChangeHandler.path)
     unregisterPartitionModificationsHandlers(partitionModificationsHandlers.keys.toSeq)
     zkClient.unregisterZNodeChildChangeHandler(topicDeletionHandler.path)
+    zkClient.unregisterZNodeChangeHandler(topicDeletionFlagHandler.path)
     // shutdown replica state machine
     replicaStateMachine.shutdown()
     zkClient.unregisterZNodeChildChangeHandler(brokerChangeHandler.path)
@@ -1378,7 +1380,7 @@ class KafkaController(val config: KafkaConfig,
       zkClient.deleteTopicDeletions(nonExistentTopics.toSeq, controllerContext.epochZkVersion)
     }
     topicsToBeDeleted --= nonExistentTopics
-    if (config.deleteTopicEnable) {
+    if (topicDeletionManager.isDeleteTopicEnabled) {
       if (topicsToBeDeleted.nonEmpty) {
         info(s"Starting topic deletion for topics ${topicsToBeDeleted.mkString(",")}")
         // mark topic ineligible for deletion if other state changes are in progress
@@ -1399,6 +1401,24 @@ class KafkaController(val config: KafkaConfig,
     }
   }
 
+  private def processTopicDeletionFlagChange(reset: Boolean = false): Unit = {
+    info("Process TopicDeletionFlagChange event")
+    if (!isActive) return
+    if (reset)
+      topicDeletionManager.resetDeleteTopicEnabled()
+    else {
+      val topicDeletionFlag = zkClient.getTopicDeletionFlag
+      if (!topicDeletionFlag.equalsIgnoreCase("true") && !topicDeletionFlag.equalsIgnoreCase("false")) {
+        info(s"Overwrite ${DeleteTopicFlagZNode.path} to ${topicDeletionManager.isDeleteTopicEnabled}")
+        zkClient.setTopicDeletionFlag(topicDeletionManager.isDeleteTopicEnabled.toString)
+      }
+      else {
+        info(s"Set isDeleteTopicEnabled flag to $topicDeletionFlag")
+        topicDeletionManager.isDeleteTopicEnabled = topicDeletionFlag.toBoolean
+      }
+    }
+  }
+
   private def processPartitionReassignment(): Unit = {
     if (!isActive) return
 
@@ -1594,6 +1614,8 @@ class KafkaController(val config: KafkaConfig,
           processPartitionModifications(topic)
         case TopicDeletion =>
           processTopicDeletion()
+        case TopicDeletionFlagChange(reset) =>
+          processTopicDeletionFlagChange(reset)
         case PartitionReassignment =>
           processPartitionReassignment()
         case PartitionReassignmentIsrChange(partition) =>
@@ -1668,6 +1690,19 @@ class TopicDeletionHandler(eventManager: ControllerEventManager) extends ZNodeCh
 
   override def handleChildChange(): Unit = eventManager.put(TopicDeletion)
 }
+/**
+  * Listener for /topic_deletion_flag znode.
+  *   If the data of the znode is set to true/false, it will trigger the in memory isDeleteTopicEnabled to be set accordingly.
+  *   If the znode data cannot be converted to boolean, it will overwrite znode with the previous valid value.
+  *   If the znode path is deleted, it will reset the in memory isDeleteTopicEnabled to the config value.
+  */
+class TopicDeletionFlagHandler(controller: KafkaController, eventManager: ControllerEventManager) extends ZNodeChangeHandler {
+  override val path: String = DeleteTopicFlagZNode.path
+
+  override def handleDataChange(): Unit = eventManager.put(TopicDeletionFlagChange())
+
+  override def handleDeletion(): Unit = eventManager.put(TopicDeletionFlagChange(true))
+}
 
 class PartitionReassignmentHandler(eventManager: ControllerEventManager) extends ZNodeChangeHandler {
   override val path: String = ReassignPartitionsZNode.path
@@ -1840,6 +1875,10 @@ case object IsrChangeNotification extends ControllerEvent {
   override def state: ControllerState = ControllerState.IsrChange
 }
 
+case class TopicDeletionFlagChange(reset: Boolean = false) extends ControllerEvent {
+  def state = ControllerState.TopicDeletionFlagChange
+}
+
 case class PreferredReplicaLeaderElection(partitionsFromAdminClientOpt: Option[Set[TopicPartition]],
                                           electionType: ElectionType = ZkTriggered,
                                           callback: ElectPreferredLeadersCallback = (_,_) => {}) extends ControllerEvent {

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

@@ -20,6 +20,7 @@ import kafka.server.KafkaConfig
 import kafka.utils.Logging
 import kafka.zk.KafkaZkClient
 import org.apache.kafka.common.TopicPartition
+import org.apache.zookeeper.KeeperException.{NoNodeException, NodeExistsException}
 
 import scala.collection.Set
 
@@ -28,6 +29,8 @@ trait DeletionClient {
   def deleteTopicDeletions(topics: Seq[String], epochZkVersion: Int): Unit
   def mutePartitionModifications(topic: String): Unit
   def sendMetadataUpdate(partitions: Set[TopicPartition]): Unit
+  def createDeleteTopicFlagPath(): Unit
+  def getTopicDeletionFlag(): String
 }
 
 class ControllerDeletionClient(controller: KafkaController, zkClient: KafkaZkClient) extends DeletionClient {
@@ -48,6 +51,14 @@ class ControllerDeletionClient(controller: KafkaController, zkClient: KafkaZkCli
   override def sendMetadataUpdate(partitions: Set[TopicPartition]): Unit = {
     controller.sendUpdateMetadataRequest(controller.controllerContext.liveOrShuttingDownBrokerIds.toSeq, partitions)
   }
+
+  override def createDeleteTopicFlagPath(): Unit = {
+    zkClient.createDeleteTopicFlagPath
+  }
+
+  override def getTopicDeletionFlag(): String = {
+    zkClient.getTopicDeletionFlag
+  }
 }
 
 /**
@@ -89,12 +100,21 @@ class TopicDeletionManager(config: KafkaConfig,
                            partitionStateMachine: PartitionStateMachine,
                            client: DeletionClient) extends Logging {
   this.logIdent = s"[Topic Deletion Manager ${config.brokerId}] "
-  val isDeleteTopicEnabled: Boolean = config.deleteTopicEnable
+  var isDeleteTopicEnabled: Boolean = config.deleteTopicEnable
+
+  // Try to create the znode for delete topic flag
+  try {
+    client.createDeleteTopicFlagPath()
+  } catch {
+    case _: NodeExistsException =>
+  }
 
   def init(initialTopicsToBeDeleted: Set[String], initialTopicsIneligibleForDeletion: Set[String]): Unit = {
     info(s"Initializing manager with initial deletions: $initialTopicsToBeDeleted, " +
       s"initial ineligible deletions: $initialTopicsIneligibleForDeletion")
 
+    isDeleteTopicEnabled = getDeleteTopicEnabled()
+
     if (isDeleteTopicEnabled) {
       controllerContext.queueTopicDeletion(initialTopicsToBeDeleted)
       controllerContext.topicsIneligibleForDeletion ++= initialTopicsIneligibleForDeletion & controllerContext.topicsToBeDeleted
@@ -355,4 +375,20 @@ class TopicDeletionManager(config: KafkaConfig,
       }
     }
   }
+
+  private def getDeleteTopicEnabled(): Boolean = {
+    try {
+      val deleteTopicFlag = client.getTopicDeletionFlag
+      if (deleteTopicFlag == null || (!deleteTopicFlag.equalsIgnoreCase("true") && !deleteTopicFlag.equalsIgnoreCase("false")))
+        isDeleteTopicEnabled
+      else deleteTopicFlag.toBoolean
+    } catch {
+      case _: NoNodeException => config.deleteTopicEnable
+    }
+  }
+
+  def resetDeleteTopicEnabled(): Unit = {
+    info("Reset isDeleteTopicEnabled flag to %s".format(config.deleteTopicEnable))
+    isDeleteTopicEnabled = config.deleteTopicEnable
+  }
 }

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -773,6 +773,37 @@ class KafkaZkClient private[zk] (zooKeeperClient: ZooKeeperClient, isSecure: Boo
     }
   }
 
+  /**
+   * Creates the delete topic flag znode.
+   * @throws KeeperException if there is an error while setting or creating the znode
+   */
+  def createDeleteTopicFlagPath(): Unit = {
+    createRecursive(DeleteTopicFlagZNode.path)
+  }
+
+  /**
+   * Get topic deletion flag in zookeeper.
+   * @return topic deletion flag in zookeeper.
+   */
+  def getTopicDeletionFlag: String = {
+    val getDataResponse = retryRequestUntilConnected(GetDataRequest(DeleteTopicFlagZNode.path))
+    getDataResponse.resultCode match {
+      case Code.OK => DeleteTopicFlagZNode.decode(getDataResponse.data)
+      case _ => throw getDataResponse.resultException.get
+    }
+  }
+
+  /**
+   * Set topic deletion flag in zookeeper.
+   */
+  def setTopicDeletionFlag(flag: String): Unit = {
+    val setDataResponse = retryRequestUntilConnected(SetDataRequest(DeleteTopicFlagZNode.path, DeleteTopicFlagZNode.encode(flag), -1))
+    setDataResponse.resultCode match {
+      case Code.OK =>
+      case _ => throw setDataResponse.resultException.get
+    }
+  }
+
   /**
    * Remove the given topics from the topics marked for deletion.
    * @param topics the topics to remove.

core/src/main/scala/kafka/zk/ZkData.scala

@@ -373,6 +373,12 @@ object DeleteTopicsTopicZNode {
   def path(topic: String) = s"${DeleteTopicsZNode.path}/$topic"
 }
 
+object DeleteTopicFlagZNode {
+  def path = "/topic_deletion_flag"
+  def encode(topicDeletionFlag: String): Array[Byte] = topicDeletionFlag.getBytes(UTF_8)
+  def decode(bytes: Array[Byte]): String = if (bytes != null) new String(bytes, UTF_8) else ""
+}
+
 object ReassignPartitionsZNode {
 
   /**

core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala

@@ -17,7 +17,7 @@
 package kafka.admin
 
 import kafka.log.Log
-import kafka.zk.{TopicPartitionZNode, ZooKeeperTestHarness}
+import kafka.zk.{DeleteTopicFlagZNode, TopicPartitionZNode, TopicZNode, ZooKeeperTestHarness}
 import kafka.utils.TestUtils
 import kafka.server.{KafkaConfig, KafkaServer}
 import org.junit.Assert._
@@ -411,6 +411,63 @@ class DeleteTopicTest extends ZooKeeperTestHarness {
     assertTrue("Leader should exist for topic test", leaderIdOpt.isDefined)
   }
 
+  @Test
+  def testDeleteTopicAfterEnableZkDeleteTopicFlag() {
+    val topicPartition = new TopicPartition("test", 0)
+    val topic = topicPartition.topic
+    servers = createTestTopicAndCluster(topic, deleteTopicEnabled = false)
+    // mark the topic for deletion
+    adminZkClient.deleteTopic("test")
+    TestUtils.waitUntilTrue(() => !zkClient.isTopicMarkedForDeletion(topic),
+      "Admin path /admin/delete_topic/%s path not deleted even if deleteTopic is disabled".format(topic))
+    // verify that topic test is untouched
+    assertTrue(servers.forall(_.getLogManager().getLog(topicPartition).isDefined))
+    // test the topic path exists
+    assertTrue("Topic path disappeared even when topic deletion is disabled", zkClient.pathExists(TopicZNode.path(topic)))
+    // topic test should have a leader
+    val leaderIdOpt = zkClient.getLeaderForPartition(new TopicPartition(topic, 0))
+    assertTrue("Leader should exist for topic test", leaderIdOpt.isDefined)
+
+    // Set TopicDeletionFlag to true in zk and try delete topic again
+    zkClient.setTopicDeletionFlag("true")
+    TestUtils.waitUntilTrue(() =>
+      try {
+        zkClient.getTopicDeletionFlag.equalsIgnoreCase("true")
+      } catch {
+        case _: Throwable => false
+      },
+      "TopicDeletionFlag is not set")
+    TestUtils.waitUntilTrue( () => getController()._1.kafkaController.topicDeletionManager.isDeleteTopicEnabled,
+      "Delete topic is not enabled")
+    // mark the topic for deletion
+    adminZkClient.deleteTopic("test")
+    TestUtils.verifyTopicDeletion(zkClient, "test", 1, servers)
+
+    // Set TopicDeletionFlag to invalid value in zk
+    zkClient.setTopicDeletionFlag("flase")
+    TestUtils.waitUntilTrue(() =>
+      try {
+        zkClient.getTopicDeletionFlag.equalsIgnoreCase("true")
+      } catch {
+        case _: Throwable => false
+      },
+      "TopicDeletionFlag is not overwritten")
+
+    // delete TopicDeletionFlagPath in zk
+    zkClient.deletePath(DeleteTopicFlagZNode.path)
+    TestUtils.waitUntilTrue(() =>
+      try {
+        !zkClient.pathExists(DeleteTopicFlagZNode.path)
+      } catch {
+        case _: Throwable => false
+      },
+      "TopicDeletionFlagPath is not deleted")
+    TestUtils.waitUntilTrue(() =>
+      getController()._1.kafkaController.topicDeletionManager.isDeleteTopicEnabled == false,
+      "Topic deletion flag is not rest"
+    )
+  }
+
   @Test
   def testDeletingPartiallyDeletedTopic() {
     /**