From 40079b47d8a2e58ea2b4559349d7ac1cf2720f69 Mon Sep 17 00:00:00 2001 From: asingh Date: Thu, 16 Jul 2015 23:12:12 -0700 Subject: [PATCH] KAFKA-2345: Attempt to delete a topic already marked for deletion throws ZkNodeExistsException --- core/src/main/scala/kafka/admin/AdminUtils.scala | 8 +++++++- .../TopicAlreadyMarkedForDeletionException.scala | 21 +++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) create mode 100644 core/src/main/scala/kafka/common/TopicAlreadyMarkedForDeletionException.scala diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala index f06edf4..2b4e028 100644 --- a/core/src/main/scala/kafka/admin/AdminUtils.scala +++ b/core/src/main/scala/kafka/admin/AdminUtils.scala @@ -163,7 +163,13 @@ object AdminUtils extends Logging { } def deleteTopic(zkClient: ZkClient, topic: String) { - ZkUtils.createPersistentPath(zkClient, ZkUtils.getDeleteTopicPath(topic)) + try { + ZkUtils.createPersistentPath(zkClient, ZkUtils.getDeleteTopicPath(topic)) + } catch { + case e1: ZkNodeExistsException => throw new TopicAlreadyMarkedForDeletionException( + "topic %s is already marked for deletion".format(topic)) + case e2: Throwable => throw new AdminOperationException(e2.toString) + } } def isConsumerGroupActive(zkClient: ZkClient, group: String) = { diff --git a/core/src/main/scala/kafka/common/TopicAlreadyMarkedForDeletionException.scala b/core/src/main/scala/kafka/common/TopicAlreadyMarkedForDeletionException.scala new file mode 100644 index 0000000..c83cea9 --- /dev/null +++ b/core/src/main/scala/kafka/common/TopicAlreadyMarkedForDeletionException.scala @@ -0,0 +1,21 @@ +/** + * 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.common + +class TopicAlreadyMarkedForDeletionException(message: String) extends RuntimeException(message) { +} \ No newline at end of file -- 2.3.2 (Apple Git-55)