MINOR: enable topic deletion in the KIP-500 controller by cmccabe · Pull Request #10184 · apache/kafka

cmccabe · 2021-02-23T01:45:13Z

Enable the new KIP-500 controller to delete topics.

Fix a bug where feature level records were not correctly replayed.

Fix a bug in TimelineHashMap#remove where the wrong type was being
returned.

rondagostino · 2021-02-23T13:20:42Z

Successfully ran the system test that was previously failing due to an inability to delete topics:

TC_PATHS="tests/kafkatest/tests/core/replica_scale_test.py::ReplicaScaleTest.test_clean_bounce" bash tests/docker/run_tests.sh

test_id:    kafkatest.tests.core.replica_scale_test.ReplicaScaleTest.test_clean_bounce.topic_count=50.partition_count=34.replication_factor=3.metadata_quorum=REMOTE_RAFT
status:     PASS
run time:   8 minutes 17.307 seconds

chia7712

overall LGTM. some minor questions are left. Please take a look.

cmccabe · 2021-02-24T20:30:00Z

rebased on trunk

chia7712

Thanks for updating code. Some minor comments are left. Please take a look.

cmccabe · 2021-02-25T22:40:51Z

rebased on trunk

cmccabe · 2021-02-26T07:16:05Z

rebased on trunk

chia7712

@cmccabe thanks for updating code. please take a look at following minor questions.

hachikuji · 2021-02-26T21:39:32Z

The error should be TOPIC_AUTHORIZATION_FAILED if the client does not have describe permission regardless of existence.

I think there are 5 cases:

name provided, topic exists, can't delete, maybe can describe => TOPIC_AUTHORIZATION_FAILED

name provided, topic doesn't exist, can't delete, can describe => UNKNOWN_TOPIC_OR_PARTITION

name provided, topic doesn't exist, can't delete, can't describe => TOPIC_AUTHORIZATION_FAILED

id provided, topic exists, can't delete, maybe can describe => TOPIC_AUTHORIZATION_FAILED

id provided, topic doesn't exists, can't delete, maybe can describe => UNKNOWN_TOPIC_ID

For case 4, are we exposing the topic exists by returning a different error (than case 5) in the case where we can't describe?

You're right, the earlier list was not quite right. I've revised this a bit. It's now:

no name or id => INVALID_REQUEST duplicate name or id => INVALID_REQUEST can't resolve topic id => UNKNOWN_TOPIC_ID can't locate topic name => UNKNOWN_TOPIC_OR_PARTITION no delete permission, no describe permission => UNKNOWN_TOPIC_ID (if topic id was provided) or UNKNOWN_TOPIC_OR_PARTITION (if topic name was provided) no delete permission, describe permission => TOPIC_AUTHORIZATION_FAILED with both name and id filled out correctly

The new code should implement this correctly...

@cmccabe Oh sorry this just got changed due to https://issues.apache.org/jira/browse/KAFKA-12394
So the case of no delete permission, no describe permission, topic ID provided is now TOPIC_AUTHORIZATION_FAILED. This may have been what you had initially.

junrao

@cmccabe : Thanks for the PR. Added a couple of comments.

junrao · 2021-02-26T23:36:34Z

Should we update brokersToIsrs too?

The ISRs should already have been updated by BrokerChangeRecords that were previously replayed.

Hmm, you mean PartitionChangeRecord? I don't see PartitionChangeRecord being generated from the topicDeletion request.

Sorry, you're right: we need to remove this from brokersToIsrs. Fixed.

junrao · 2021-02-26T23:42:30Z

I guess we haven't hooked up the logic to trigger the deletion of the replicas of the deleted topic in the broker?

We haven't hooked that up yet, correct. But that logic is in BrokerMetadataListener. It would probably be better to have a separate PR for that.

junrao

@cmccabe : Thanks for the explanation. A couple of more comments.

junrao · 2021-02-27T02:04:46Z

Hmm, you mean PartitionChangeRecord? I don't see PartitionChangeRecord being generated from the topicDeletion request.

junrao · 2021-02-27T02:05:27Z

We also need to delete the configuration associated with the topic.

Is the deletion of topic configurations implicit based on the DeleteTopic record? I know we discussed this, but I'm unsure what the final outcome was. I don't see any logic for this in the broker listener, but the implementation looks incomplete anyway.

Yes, it should be implicit based on the DeleteTopic record. I will fix the controller to do the right thing here. We'll also need to have the broker do that too.

chia7712

@cmccabe Thanks for updating PR.

this code ends up being complicated. Could it be separated and then reused by ControlApis and KafkaApis? I left same comment on #10223 (#10223 (comment))

ijuma · 2021-02-27T15:23:31Z

@cmccabe I pushed a commit that removes a bunch of unnecessary asScala conversions (even though these don't copy the collection, they add a shallow allocation and indirection and the code is shorter with these changes - win/win).

chia7712

@cmccabe Thanks for handling this very, very, very complicated code. a couple of comments are left. Please take a look.

hachikuji · 2021-03-01T22:23:05Z

Is the deletion of topic configurations implicit based on the DeleteTopic record? I know we discussed this, but I'm unsure what the final outcome was. I don't see any logic for this in the broker listener, but the implementation looks incomplete anyway.

rondagostino · 2021-03-02T21:48:01Z

The below, if added as tests/kafkatest/sanity_checks/test_delete_topic.py, fails for the Raft cases on this PR branch as of this moment because the broker fails to shutdown. The following appears in the controller log:

[2021-03-02 21:41:13,354] INFO [Controller 1] Unfenced broker 1 has requested and been granted a controlled shutdown. (org.apache.kafka.controller.BrokerHeartbeatManager)
[2021-03-02 21:41:13,355] WARN [Controller 1] org.apache.kafka.controller.QuorumController@3fa533f1: failed with unknown server exception RuntimeException at epoch 1 in 802 us.  Reverting to last committed offset 5. (org.apache.kafka.controller.QuorumController)
java.lang.RuntimeException: Topic ID VnD54LHq2t3qq_m1WLasZg existed in isrMembers, but not in the topics map.
	at org.apache.kafka.controller.ReplicationControlManager.handleNodeDeactivated(ReplicationControlManager.java:752)
	at org.apache.kafka.controller.ReplicationControlManager.processBrokerHeartbeat(ReplicationControlManager.java:931)
	at org.apache.kafka.controller.QuorumController$1.generateRecordsAndResult(QuorumController.java:911)
	at org.apache.kafka.controller.QuorumController$ControllerWriteEvent.run(QuorumController.java:419)
	at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
	at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
	at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
	at java.lang.Thread.run(Thread.java:748)

Maybe add this system test to this PR as tests/kafkatest/sanity_checks/test_delete_topic.py?

# 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.


from ducktape.mark import matrix
from ducktape.mark.resource import cluster
from ducktape.tests.test import Test

from kafkatest.services.kafka import KafkaService, quorum
from kafkatest.services.zookeeper import ZookeeperService
import time

class TestDeleteTopic(Test):
    """Sanity checks that we can create and delete a topic and then shutdown."""
    def __init__(self, test_context):
        super(TestDeleteTopic, self).__init__(test_context)

        self.topic = "test_topic"
        self.zk = ZookeeperService(test_context, num_nodes=1) if quorum.for_test(test_context) == quorum.zk else None
        self.kafka = KafkaService(test_context, num_nodes=1, zk=self.zk,
                                  topics={self.topic: {"partitions": 1, "replication-factor": 1}},
                                  controller_num_nodes_override=1)
    def setUp(self):
        if self.zk:
            self.zk.start()

    @cluster(num_nodes=2)
    @matrix(metadata_quorum=quorum.all)
    def test_delete_topic(self, metadata_quorum):
        """
        Test that we can create and delete a topic and then shutdown
        """
        self.kafka.start()
        self.kafka.delete_topic(self.topic)
        time.sleep(10) # give it a bit to take effect
        self.kafka.stop() # explicit stop so that failure to stop fails the test

hachikuji · 2021-03-02T22:07:27Z

@rondagostino I believe that error will be fixed by https://issues.apache.org/jira/browse/KAFKA-12403.

cmccabe · 2021-03-02T22:12:03Z

Thanks for the reviews! I reworked the authentication, validation, and de-duplication code a lot. The new logic should take into account the issues pointed out here. I resolved a few comment threads since they refer to code that was refactored-- please take another look if you get a chance.

To clarify a bit, RemoveTopicRecord should imply some other effects:

All topic configs for the affected topic should be deleted
We should delete all the partitions of the deleted topic
We should remove the topic from brokersToIsrs

The fact that it wasn't doing these things was a bug... it's fixed now. This should also allow the ducktape test to work (cc @rondagostino )

We also have a JIRA to follow up on the broker side: https://issues.apache.org/jira/browse/KAFKA-12403

Enable the new KIP-500 controller to delete topics. Fix a bug where feature level records were not correctly replayed. Fix a bug in TimelineHashMap#remove where the wrong type was being returned.

rondagostino · 2021-03-03T01:20:27Z

[2021-03-02T22:42:17.438Z] [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-10184/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:76:8: Unused import - java.util.Set. [UnusedImports]

chia7712

@cmccabe thanks for updating code.

The following comments are used to make sure the error handle (and response) is consistent to #10223

please take a look.

hachikuji

LGTM. @junrao Would you mind checking over the last two commits I added?

junrao

@hachikuji : Thanks for the updated PR. Just a couple of more comments.

junrao · 2021-03-03T23:46:16Z

+                brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]);
+            }
+        }
+        brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER);


Hmm, why do we need to remove for -1 broker? It doesn't seem that brokersToIsrs tracks that.

The test case BrokersToIsrsTest.testNoLeader suggests that it is a possible case. It looks like the path through ReplicationControlManager.handleNodeDeactivated could result in a PartitionChangeRecord which has leaderId set to -1.

It's true and a partition could have isr and no leader. However, in that case, isrMembers in brokersToIsrs will still be updated with key from replicaId in isr and isr will never have -1 in its list. The noLeader info is only stored in the value of isrMembers.

I stepped through testNoLeader and it seems that -1 can indeed be a key in isrMembers. The noLeaderIterator makes the expectation explicit.

Got it. The following comment confirmed this.

/** * A map of broker IDs to the partitions that the broker is in the ISR for. * Partitions with no isr members appear in this map under id NO_LEADER. */ private final TimelineHashMap<Integer, TimelineHashMap<Uuid, int[]>> isrMembers;

chia7712

+1 to this nice patch. Some trivial comments are left. Please take a look.

chia7712 · 2021-03-04T04:06:48Z

+    val barId = Uuid.fromString("VlFu5c51ToiNx64wtwkhQw")
+    val controller = new MockController.Builder().
+      newInitialTopic("foo", fooId).build()
+    controller.setActive(false)


this controller is mock so disabling active works well for this test. However, I did not observe the check of control activity in production code. Could you share that with me?

See QuorumController.QuorumMetaLogListener for the callbacks that make the controller active or inactive.

junrao

@hachikuji : Thanks for the updated PR. LGTM too.

junrao · 2021-03-04T05:20:50Z

+                brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]);
+            }
+        }
+        brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER);


Got it. The following comment confirmed this.

/** * A map of broker IDs to the partitions that the broker is in the ISR for. * Partitions with no isr members appear in this map under id NO_LEADER. */ private final TimelineHashMap<Integer, TimelineHashMap<Uuid, int[]>> isrMembers;

hachikuji · 2021-03-04T19:15:29Z

Thanks for reviews. I will merge this on behalf of @cmccabe to trunk and 2.8.

chia7712 · 2021-03-04T19:44:57Z

@hachikuji this PR breaks the build (this PR was merged after #10253). Could you file a hot-fix?

This patch enables delete topic support for the new KIP-500 controller. Also fixes the following: - Fix a bug where feature level records were not correctly replayed. - Fix a bug in TimelineHashMap#remove where the wrong type was being returned. Reviewers: Jason Gustafson <jason@confluent.io>, Justine Olshan <jolshan@confluent.io>, Ron Dagostino <rdagostino@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>, Jun Rao <junrao@gmail.com> Co-authored-by: Jason Gustafson <jason@confluent.io>

Topic deletions should be atomic. This fixes a build error caused by merging of both #10253 and #10184 at about the same time. Reviewers: David Arthur <mumrah@gmail.com>

cmccabe added the kraft label Feb 23, 2021

ijuma reviewed Feb 23, 2021

View reviewed changes

Comment thread config/log4j.properties Outdated

chia7712 reviewed Feb 23, 2021

View reviewed changes

ijuma reviewed Feb 23, 2021

View reviewed changes

cmccabe force-pushed the kip500_delete branch from b05a027 to f31fb91 Compare February 24, 2021 20:29

chia7712 reviewed Feb 25, 2021

View reviewed changes

cmccabe force-pushed the kip500_delete branch from 7822e76 to 08aedff Compare February 25, 2021 22:40

cmccabe force-pushed the kip500_delete branch from 08aedff to b56ea48 Compare February 26, 2021 07:15

chia7712 reviewed Feb 26, 2021

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

Comment thread metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java Outdated

Comment thread metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java Outdated

cmccabe force-pushed the kip500_delete branch from b56ea48 to 3c649c4 Compare February 26, 2021 20:31

hachikuji reviewed Feb 26, 2021

View reviewed changes

junrao reviewed Feb 26, 2021

View reviewed changes

cmccabe force-pushed the kip500_delete branch from f571a4c to ac03b21 Compare February 27, 2021 01:20

junrao reviewed Feb 27, 2021

View reviewed changes

chia7712 mentioned this pull request Feb 27, 2021

KAFKA-12394: Return TOPIC_AUTHORIZATION_FAILED in delete topic response if no describe permission #10223

Merged

3 tasks

chia7712 reviewed Feb 27, 2021

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

chia7712 reviewed Mar 1, 2021

View reviewed changes

hachikuji reviewed Mar 1, 2021

View reviewed changes

cmccabe force-pushed the kip500_delete branch 2 times, most recently from 0274964 to 9424061 Compare March 2, 2021 20:30

MINOR: enable topic deletion in the KIP-500 controller

f020e94

Enable the new KIP-500 controller to delete topics. Fix a bug where feature level records were not correctly replayed. Fix a bug in TimelineHashMap#remove where the wrong type was being returned.

cmccabe force-pushed the kip500_delete branch from 9424061 to ddb6b87 Compare March 2, 2021 22:40

cmccabe force-pushed the kip500_delete branch from ddb6b87 to f020e94 Compare March 2, 2021 22:41

chia7712 reviewed Mar 3, 2021

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

Comment thread core/src/main/scala/kafka/server/ControllerApis.scala Outdated

hachikuji reviewed Mar 3, 2021

View reviewed changes

Comment thread metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java Outdated

Comment thread metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java Outdated

hachikuji added 2 commits March 3, 2021 13:05

Ensure topic config removed only during replay

f71043e

Consistent error handling with KafkaApis

746f0bb

hachikuji force-pushed the kip500_delete branch from c711097 to 746f0bb Compare March 3, 2021 22:30

hachikuji approved these changes Mar 3, 2021

View reviewed changes

junrao reviewed Mar 3, 2021

View reviewed changes

Fix ControllerApisTest package name

654684d

chia7712 approved these changes Mar 4, 2021

View reviewed changes

junrao approved these changes Mar 4, 2021

View reviewed changes

Final cleanups

6aec8bd

hachikuji merged commit eebc6f2 into apache:trunk Mar 4, 2021

hachikuji mentioned this pull request Mar 4, 2021

HOTFIX: Controller topic deletion should be atomic #10264

Merged

3 tasks

Conversation

cmccabe commented Feb 23, 2021

Uh oh!

Uh oh!

rondagostino commented Feb 23, 2021

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cmccabe commented Feb 24, 2021

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cmccabe commented Feb 25, 2021

Uh oh!

cmccabe commented Feb 26, 2021

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jolshan Feb 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jolshan Mar 2, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jolshan Feb 27, 2021 •

edited

Loading

jolshan Mar 2, 2021 •

edited

Loading