From df8ad7e1391837203f052a84134e32855accfd26 Mon Sep 17 00:00:00 2001 From: Jon Bringhurst Date: Fri, 11 Sep 2026 15:38:35 -0700 Subject: [PATCH 1/2] kafka: resume interrupted deletion without restarting orphaned reassignment --- .../kafka/controller/KafkaController.scala | 22 +++++ .../BridgeInterruptedDeletionTest.scala | 99 +++++++++++++++++++ 2 files changed, 121 insertions(+) create mode 100644 core/src/test/scala/unit/kafka/controller/BridgeInterruptedDeletionTest.scala diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index 1411bbb46dd51..888f8d84b5c8a 100644 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/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(",")}") diff --git a/core/src/test/scala/unit/kafka/controller/BridgeInterruptedDeletionTest.scala b/core/src/test/scala/unit/kafka/controller/BridgeInterruptedDeletionTest.scala new file mode 100644 index 0000000000000..602c539f10347 --- /dev/null +++ b/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() + } +} From 8086d1796816bd2383844b7b9f9b05ebd8b33c9b Mon Sep 17 00:00:00 2001 From: Jon Bringhurst Date: Fri, 11 Sep 2026 16:16:22 -0700 Subject: [PATCH 2/2] kafka: echo gated deletion intent in native StopReplica responses --- .../main/scala/kafka/server/KafkaApis.scala | 3 ++ .../unit/kafka/server/KafkaApisTest.scala | 36 ++++++++++++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 9fb4a74cd416e..5bec9a2603865 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/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) diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index 94448d2483278..7fc9dbcb0fd05 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/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