KAFKA-12403; Ensure local state deleted on `RemoveTopicRecord` received #10252

hachikuji · 2021-03-03T04:55:46Z

This patch implements additional handling logic for RemoveTopic records:

Update MetadataPartitions to ensure addition of deleted partitions to localRemoved set
Ensure topic configs are removed from ConfigRepository
Propagate deleted partitions to GroupCoordinator so that corresponding offset commits can be removed

This patch also changes the controller topic id generation logic to use Uuid.randomUuid rather than Random.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712

@hachikuji Thanks for this fix. a couple of comments. Please take a look.

core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala

core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala

ijuma · 2021-03-04T05:13:49Z

core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala

-        record.asInstanceOf[RemoveTopicRecord])
-      case QUOTA_RECORD => handleQuotaRecord(imageBuilder,
-        record.asInstanceOf[QuotaRecord])
-      // TODO: handle FEATURE_LEVEL_RECORD


Is this TODO no longer relevant?

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

chia7712

@hachikuji thanks for your response and updates. Left some questions for this PR.

chia7712 · 2021-03-04T06:12:40Z

core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala

    }
    newNameMap.put(partition.topicName, newPartitionMap)
  }

+  private def maybeAddToLocalRemoved(partition: MetadataPartition): Unit = {
+    if (partition.isReplicaFor(brokerId)) {
+      val currentTopicId = newReverseIdMap.get(partition.topicName)


Is it possible that newReverseIdMap has no related id? For example, PartitionRecord is processed before TopicRecord or TopicRecord was discarded (due to error)?

For another, could it be replaced by prevPartitions.contains(partition.topicName)? It seems all we want to check is the existence of topic name in previous image.

The intent is to only return the change in _localRemoved if the topic existed in the previous image. If we only check topic name, then successive deletions and recreations might leave some partitions in _localRemoved that were not in the previous image.

It's worth noting that this is strictly more defensive than the current replay logic requires. A new image is built for each batch of records from the controller, and we would never see a topic deleted and recreated (or vice versa) in the same batch. This is an implicit contract though and not protected by the builder API, so I thought we might as well try to make the logic more resilient.

Thanks for nice explanation. I have another question for this check. It seems to me three collections in prevPartitions should be consistent. For example: a topic which exists one of collection should also exists in other two (vice versa). If above comment is right, why we need this if-else? Calling prevPartitions.contains(partition.topicName) appears to be enough?

core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala

core/src/test/scala/kafka/server/metadata/MetadataPartitionsTest.scala

hachikuji · 2021-03-04T23:29:20Z

I am working on some integration tests here. Hopefully then we can wrap this up.

hachikuji · 2021-03-04T23:55:06Z

Well, I was going to write some integration tests, but it seems we are still awaiting some infrastructure for that. I tested it out manually and it works correctly. I see the brokers deleting the partition data as expected. I tested deletion as well as recreation.

chia7712

@hachikuji thanks for updating code. a couple of comments are left. Otherwise, LGTM.

core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala

chia7712

thanks for this nice patch. LGTM. one trivial question is left.

chia7712 · 2021-03-08T17:43:31Z

core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala

+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+class BrokerMetadataListenerTest {


It seems not all messages are covered in this class? If so, is there a jira to complete those test cases?

Good idea. I filed https://issues.apache.org/jira/browse/KAFKA-12437.

…ed (#10252) This patch implements additional handling logic for `RemoveTopic` records: - Update `MetadataPartitions` to ensure addition of deleted partitions to `localRemoved` set - Ensure topic configs are removed from `ConfigRepository` - Propagate deleted partitions to `GroupCoordinator` so that corresponding offset commits can be removed This patch also changes the controller topic id generation logic to use `Uuid.randomUuid` rather than `Random`. Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>

KAFKA-12403; Ensure local state deleted on RemoveTopicRecord received

d0a0c65

chia7712 reviewed Mar 3, 2021

View reviewed changes

hachikuji added 2 commits March 3, 2021 11:55

Various cleanups

1b6eec4

Fix 2.12 compilation error

2bec718

ijuma reviewed Mar 4, 2021

View reviewed changes

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java Show resolved Hide resolved

chia7712 reviewed Mar 4, 2021

View reviewed changes

hachikuji added 3 commits March 4, 2021 12:55

Merge remote-tracking branch 'upstream/trunk' into KAFKA-12403

9043cd9

Additional test assertions

89642e6

Merge remote-tracking branch 'upstream/trunk' into KAFKA-12403

851e9c3

chia7712 reviewed Mar 6, 2021

View reviewed changes

core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala Outdated Show resolved Hide resolved

core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala Show resolved Hide resolved

core/src/main/scala/kafka/server/metadata/MetadataPartitions.scala Show resolved Hide resolved

Simplify not equals check

85c26fa

chia7712 approved these changes Mar 8, 2021

View reviewed changes

hachikuji merged commit 31a121c into apache:trunk Mar 8, 2021

ijuma added the kraft label Aug 11, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-12403; Ensure local state deleted on `RemoveTopicRecord` received #10252

KAFKA-12403; Ensure local state deleted on `RemoveTopicRecord` received #10252

hachikuji commented Mar 3, 2021 •

edited

chia7712 left a comment

ijuma Mar 4, 2021

chia7712 left a comment

chia7712 Mar 4, 2021

chia7712 Mar 4, 2021

hachikuji Mar 4, 2021

chia7712 Mar 6, 2021

hachikuji commented Mar 4, 2021

hachikuji commented Mar 4, 2021

chia7712 left a comment

chia7712 left a comment

chia7712 Mar 8, 2021

hachikuji Mar 8, 2021

KAFKA-12403; Ensure local state deleted on RemoveTopicRecord received #10252

KAFKA-12403; Ensure local state deleted on RemoveTopicRecord received #10252

Conversation

hachikuji commented Mar 3, 2021 • edited

Committer Checklist (excluded from commit message)

chia7712 left a comment

Choose a reason for hiding this comment

ijuma Mar 4, 2021

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 Mar 4, 2021

Choose a reason for hiding this comment

chia7712 Mar 4, 2021

Choose a reason for hiding this comment

hachikuji Mar 4, 2021

Choose a reason for hiding this comment

chia7712 Mar 6, 2021

Choose a reason for hiding this comment

hachikuji commented Mar 4, 2021

hachikuji commented Mar 4, 2021

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 Mar 8, 2021

Choose a reason for hiding this comment

hachikuji Mar 8, 2021

Choose a reason for hiding this comment

KAFKA-12403; Ensure local state deleted on `RemoveTopicRecord` received #10252

KAFKA-12403; Ensure local state deleted on `RemoveTopicRecord` received #10252

hachikuji commented Mar 3, 2021 •

edited