Recover interrupted topic deletion on the 3.0 bridge

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

@@ -1175,8 +1175,30 @@ class KafkaController(val config: KafkaConfig,
     }
   }
 
+  private[controller] def recoverInterruptedTopicDeletions(topicsToBeDeleted: Set[String]): Unit = {
+    if (!config.liProtocolBridgeTopicDeletionStateCleanupActive || !topicDeletionManager.isDeleteTopicEnabled) return
+    // Preserve every assigned replica for a marked deletion whose leader/ISR znode
+    // disappeared during recursive cleanup. Resuming its reassignment cannot succeed.
+    val interrupted = controllerContext.partitionsBeingReassigned.iterator.filter { tp =>
+      topicsToBeDeleted.contains(tp.topic) && controllerContext.partitionLeadershipInfo(tp).isEmpty
+    }.toVector
+    interrupted.foreach { tp =>
+      val assignment = ReplicaAssignment(controllerContext.partitionReplicaAssignment(tp))
+      val topicAssignments = controllerContext.partitionFullReplicaAssignmentForTopic(tp.topic) + (tp -> assignment)
+      zkClient.setTopicAssignment(tp.topic, controllerContext.topicIds.get(tp.topic),
+        topicAssignments.toMap, controllerContext.epochZkVersion)
+      controllerContext.updatePartitionFullReplicaAssignment(tp, assignment)
+    }
+    if (interrupted.nonEmpty) {
+      val interruptedSet = interrupted.toSet
+      maybeRemoveFromZkReassignment((tp, _) => interruptedSet.contains(tp))
+      controllerContext.partitionsBeingReassigned --= interrupted
+    }
+  }
+
   private def fetchTopicDeletionsInProgress(): (Set[String], Set[String]) = {
     val topicsToBeDeleted = zkClient.getTopicDeletions.toSet
+    recoverInterruptedTopicDeletions(topicsToBeDeleted)
     val topicsForWhichPartitionReassignmentIsInProgress = controllerContext.partitionsBeingReassigned.map(_.topic)
     val topicsIneligibleForDeletion = topicsForWhichPartitionReassignmentIsInProgress
     info(s"List of topics to be deleted: ${topicsToBeDeleted.mkString(",")}")

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -370,6 +370,9 @@ class KafkaApis(val requestChannel: RequestChannel,
         .setTopicName(tp.topic)
         .setPartitionIndex(tp.partition)
         .setErrorCode(error.code)
+        // Native v4 callbacks use this bit even when the request was not combined.
+        .setDeletePartition(config.liProtocolBridgeTopicDeletionStateCleanupActive &&
+          stopReplicaRequest.version >= 4 && partitionStates(tp).deletePartition())
 
     new StopReplicaResponse(new StopReplicaResponseData()
       .setErrorCode(error.code)

core/src/test/scala/unit/kafka/controller/BridgeInterruptedDeletionTest.scala

@@ -0,0 +1,99 @@
+/*
+ * 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.controller
+
+import kafka.api.LeaderAndIsr
+import kafka.server.{BrokerFeatures, DelegationTokenManager, FinalizedFeatureCache, KafkaConfig}
+import kafka.utils.TestUtils
+import kafka.zk.{BrokerInfo, KafkaZkClient}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.ControllerMovedException
+import org.apache.kafka.common.metrics.Metrics
+import org.apache.kafka.common.utils.MockTime
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+import org.mockito.ArgumentMatchers.{any, anyInt, anyString}
+import org.mockito.Mockito._
+
+import scala.collection.mutable.ArrayBuffer
+
+class BridgeInterruptedDeletionTest {
+  private def controller(client: KafkaZkClient, cleanup: Boolean, deletion: Boolean): KafkaController = {
+    val props = TestUtils.createBrokerConfig(1, "zkConnect")
+    props.put(KafkaConfig.LiProtocolBridgeTopicDeletionStateCleanupEnableProp, cleanup.toString)
+    props.put(KafkaConfig.DeleteTopicEnableProp, deletion.toString)
+    when(client.getTopicDeletionFlag).thenReturn(deletion.toString)
+    new KafkaController(KafkaConfig.fromProps(props), ArrayBuffer(client), new MockTime(), mock(classOf[Metrics]),
+      mock(classOf[BrokerInfo]), 0L, mock(classOf[DelegationTokenManager]), mock(classOf[BrokerFeatures]),
+      mock(classOf[FinalizedFeatureCache]))
+  }
+
+  @Test
+  def testRecoveryRetainsReplicasAndUnrelatedReassignments(): Unit = {
+    for (cleanup <- Seq(false, true); deletion <- Seq(false, true)) {
+      val client = mock(classOf[KafkaZkClient])
+      val broker = controller(client, cleanup, deletion)
+      val context = broker.controllerContext
+      val deleted = new TopicPartition("deleted", 0)
+      val live = new TopicPartition("live", 0)
+      val withState = new TopicPartition("with-state", 0)
+      val assignment = ReplicaAssignment(Seq(1, 2), Seq(2), Seq(1))
+      Seq(deleted, live, withState).foreach { tp =>
+        context.updatePartitionFullReplicaAssignment(tp, assignment)
+        context.partitionsBeingReassigned.add(tp)
+      }
+      context.putPartitionLeadershipInfo(withState, LeaderIsrAndControllerEpoch(LeaderAndIsr(1, List(1, 2)), 0))
+      when(client.reassignPartitionsInProgress).thenReturn(true)
+      when(client.getPartitionReassignment).thenReturn(Map(deleted -> Seq(2), live -> Seq(2)))
+      try {
+        broker.recoverInterruptedTopicDeletions(Set("deleted", "with-state"))
+        assertEquals(assignment, context.partitionFullReplicaAssignment(live))
+        assertEquals(assignment, context.partitionFullReplicaAssignment(withState))
+        assertTrue(context.partitionsBeingReassigned.contains(live))
+        assertTrue(context.partitionsBeingReassigned.contains(withState))
+        if (cleanup && deletion) {
+          val retained = ReplicaAssignment(Seq(1, 2))
+          assertEquals(retained, context.partitionFullReplicaAssignment(deleted))
+          assertEquals(Set(live, withState), context.partitionsBeingReassigned.toSet)
+          verify(client).setTopicAssignment("deleted", None, Map(deleted -> retained), context.epochZkVersion)
+          verify(client).setOrCreatePartitionReassignment(Map(live -> Seq(2)), context.epochZkVersion)
+        } else {
+          assertEquals(assignment, context.partitionFullReplicaAssignment(deleted))
+          verify(client, never()).setTopicAssignment(anyString(), any(), any(), anyInt())
+        }
+      } finally broker.shutdown()
+    }
+  }
+
+  @Test
+  def testWriteFailureDoesNotPublishNewAssignment(): Unit = {
+    val client = mock(classOf[KafkaZkClient])
+    val broker = controller(client, cleanup = true, deletion = true)
+    val tp = new TopicPartition("deleted", 0)
+    val assignment = ReplicaAssignment(Seq(1, 2), Seq(2), Seq(1))
+    broker.controllerContext.updatePartitionFullReplicaAssignment(tp, assignment)
+    broker.controllerContext.partitionsBeingReassigned.add(tp)
+    val failure = new ControllerMovedException("fenced")
+    doAnswer(_ => throw failure).when(client).setTopicAssignment(anyString(), any(), any(), anyInt())
+    try {
+      assertSame(failure, assertThrows(classOf[ControllerMovedException],
+        () => broker.recoverInterruptedTopicDeletions(Set("deleted"))))
+      assertEquals(assignment, broker.controllerContext.partitionFullReplicaAssignment(tp))
+      assertTrue(broker.controllerContext.partitionsBeingReassigned.contains(tp))
+    } finally broker.shutdown()
+  }
+}

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -55,7 +55,7 @@ import org.apache.kafka.common.message.UpdateMetadataRequestData.{UpdateMetadata
 import org.apache.kafka.common.message._
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.{ClientInformation, ListenerName}
-import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, MessageUtil}
 import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity}
 import org.apache.kafka.common.record.FileRecords.TimestampAndOffset
 import org.apache.kafka.common.record._
@@ -3049,6 +3049,40 @@ class KafkaApisTest {
     EasyMock.verify(replicaManager)
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = Array(false, true))
+  def testNativeStopReplicaResponseIdentifiesDeletionWithCleanup(cleanup: Boolean): Unit = {
+    val states = (0 to 2).map { partition =>
+      new StopReplicaPartitionState().setPartitionIndex(partition).setLeaderEpoch(1)
+        .setDeletePartition(partition != 2)
+    }
+    val stopRequest = new StopReplicaRequest.Builder(4.toShort, 0, 5, 10L, 10L, false,
+      Seq(new StopReplicaTopicState().setTopicName("foo").setPartitionStates(states.asJava)).asJava).build()
+    val request = buildRequest(stopRequest)
+    val errors = mutable.Map(new TopicPartition("foo", 0) -> Errors.NONE,
+      new TopicPartition("foo", 1) -> Errors.FENCED_LEADER_EPOCH,
+      new TopicPartition("foo", 2) -> Errors.FENCED_LEADER_EPOCH)
+    val capturedResponse: Capture[AbstractResponse] = EasyMock.newCapture()
+    EasyMock.expect(controller.brokerEpoch).andStubReturn(10L)
+    EasyMock.expect(replicaManager.stopReplicas(request.context.correlationId, 0, 5,
+      stopRequest.partitionStates().asScala)).andReturn((errors, Errors.NONE))
+    EasyMock.expect(requestChannel.sendResponse(EasyMock.eq(request),
+      EasyMock.capture(capturedResponse), EasyMock.eq(None)))
+    EasyMock.replay(controller, replicaManager, requestChannel)
+
+    createKafkaApis(overrideProperties = Map(
+      KafkaConfig.LiProtocolBridgeTopicDeletionStateCleanupEnableProp -> cleanup.toString))
+      .handleStopReplicaRequest(request)
+    val response = capturedResponse.getValue.asInstanceOf[StopReplicaResponse]
+    val decoded = StopReplicaResponse.parse(MessageUtil.toByteBuffer(response.data(), 4.toShort), 4.toShort)
+    assertEquals(3, decoded.partitionErrors().size())
+    decoded.partitionErrors().asScala.foreach { partition =>
+      assertEquals(cleanup && partition.partitionIndex() != 2, partition.deletePartition())
+      assertEquals(errors(new TopicPartition("foo", partition.partitionIndex())).code(), partition.errorCode())
+    }
+    EasyMock.verify(replicaManager, requestChannel)
+  }
+
   @Test
   def testStopReplicaRequestWithCurrentBrokerEpoch(): Unit = {
     val currentBrokerEpoch = 1239875L