Skip to content

KAFKA-9946; StopReplicaRequest deletePartition changes may cause premature topic deletion handling in controller #8609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 2 commits into from
May 6, 2020
Merged

KAFKA-9946; StopReplicaRequest deletePartition changes may cause premature topic deletion handling in controller #8609

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e2797e2
KAFKA-9946; StopReplicaRequest deletePartition changes may cause prem…
dajac May 4, 2020
272e333
address feedback
dajac May 5, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 19 additions & 31 deletions core/src/main/scala/kafka/controller/ControllerChannelManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -550,6 +550,22 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
else if (config.interBrokerProtocolVersion >= KAFKA_2_2_IV0) 1
else 0

def responseCallback(brokerId: Int, isPartitionDeleted: TopicPartition => Boolean)
(response: AbstractResponse): Unit = {
val stopReplicaResponse = response.asInstanceOf[StopReplicaResponse]
val partitionErrorsForDeletingTopics = mutable.Map.empty[TopicPartition, Errors]
stopReplicaResponse.partitionErrors.asScala.foreach { pe =>
val tp = new TopicPartition(pe.topicName, pe.partitionIndex)
if (controllerContext.isTopicDeletionInProgress(pe.topicName) &&
isPartitionDeleted(tp)) {
partitionErrorsForDeletingTopics += tp -> Errors.forCode(pe.errorCode)

@dhruvilshah3 dhruvilshah3 May 5, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: partitionErrorsForDeletingTopics seems a bit ambiguous and makes it sound like it only includes partitions for which StopReplicaRequest failed. Perhaps something like partitionToError is better?

@hachikuji hachikuji May 5, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the name seems ok. To me it means that the map includes the errors of all topic being deleted. It might be nice if it could reflect that this is only covering partitions which were also requested to be deleted in the StopReplica request, but that name probably becomes unwieldy.

}
}
if (partitionErrorsForDeletingTopics.nonEmpty)
sendEvent(TopicDeletionStopReplicaResponseReceived(brokerId, stopReplicaResponse.error,
partitionErrorsForDeletingTopics))
}

stopReplicaRequestMap.foreach { case (brokerId, partitionStates) =>
if (controllerContext.liveOrShuttingDownBrokerIds.contains(brokerId)) {
if (traceEnabled)
Expand All @@ -572,23 +588,8 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
val stopReplicaRequestBuilder = new StopReplicaRequest.Builder(
stopReplicaRequestVersion, controllerId, controllerEpoch, brokerEpoch,
false, stopReplicaTopicState.values.toBuffer.asJava)

sendRequest(brokerId, stopReplicaRequestBuilder, (r: AbstractResponse) => {
val stopReplicaResponse = r.asInstanceOf[StopReplicaResponse]
val partitionErrorsForDeletingTopics = mutable.Map.empty[TopicPartition, Errors]
stopReplicaResponse.partitionErrors.asScala.foreach { pe =>
val tp = new TopicPartition(pe.topicName, pe.partitionIndex)
// Verify that the topic deletion is in progress and
// that the request deleted the replica
if (controllerContext.isTopicDeletionInProgress(pe.topicName) &&
partitionStates.get(tp).exists(_.deletePartition)) {
partitionErrorsForDeletingTopics += tp -> Errors.forCode(pe.errorCode)
}
}
if (partitionErrorsForDeletingTopics.nonEmpty)
sendEvent(TopicDeletionStopReplicaResponseReceived(brokerId, stopReplicaResponse.error,
partitionErrorsForDeletingTopics))
})
sendRequest(brokerId, stopReplicaRequestBuilder,
responseCallback(brokerId, tp => partitionStates.get(tp).exists(_.deletePartition)))
} else {
var numPartitionStateWithDelete = 0
var numPartitionStateWithoutDelete = 0
Expand All @@ -614,20 +615,7 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig,
val stopReplicaRequestBuilder = new StopReplicaRequest.Builder(
stopReplicaRequestVersion, controllerId, controllerEpoch, brokerEpoch,
true, topicStatesWithDelete.values.toBuffer.asJava)

sendRequest(brokerId, stopReplicaRequestBuilder, (r: AbstractResponse) => {
val stopReplicaResponse = r.asInstanceOf[StopReplicaResponse]
val partitionErrorsForDeletingTopics = mutable.Map.empty[TopicPartition, Errors]
stopReplicaResponse.partitionErrors.asScala.foreach { pe =>
if (controllerContext.isTopicDeletionInProgress(pe.topicName)) {
partitionErrorsForDeletingTopics +=
new TopicPartition(pe.topicName, pe.partitionIndex) -> Errors.forCode(pe.errorCode)
}
}
if (partitionErrorsForDeletingTopics.nonEmpty)
sendEvent(TopicDeletionStopReplicaResponseReceived(brokerId, stopReplicaResponse.error,
partitionErrorsForDeletingTopics))
})
sendRequest(brokerId, stopReplicaRequestBuilder, responseCallback(brokerId, _ => true))
}

if (topicStatesWithoutDelete.nonEmpty) {
Expand Down