Mailing-List: contact dev-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@kafka.apache.org
Date: Sun, 3 Dec 2017 17:02:00 +0000 (UTC)
From: "kic (JIRA)" <jira@apache.org>
To: dev@kafka.apache.org
Message-ID: <JIRA.13122448.1512320487000.374087.1512320520880@Atlassian.JIRA>
In-Reply-To: <JIRA.13122448.1512320487000@Atlassian.JIRA>
References: <JIRA.13122448.1512320487000@Atlassian.JIRA> <JIRA.13122448.1512320487453@jira-lw-us.apache.org>
Subject: [jira] [Created] (KAFKA-6302) Topic can not be recreated after it
 is deleted
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Sun, 03 Dec 2017 17:02:07 -0000

kic created KAFKA-6302:
--------------------------

             Summary: Topic can not be recreated after it is deleted
                 Key: KAFKA-6302
                 URL: https://issues.apache.org/jira/browse/KAFKA-6302
             Project: Kafka
          Issue Type: Bug
          Components: admin, clients
    Affects Versions: 1.0.0
            Reporter: kic


I use an embedded kafka for unit test. My application relies on the ability to recreate topics programmatically. Currently it is not possible to re-create a topic after it has been deleted.

{code}
// needs compile time depedency 'net.manub:scalatest-embedded-kafka_2.11:1.0.0'
package kic.kafka.embedded

import java.util.Properties

import org.apache.kafka.clients.admin.{AdminClient, NewTopic}
import org.scalatest._

import scala.collection.JavaConverters._

class EmbeddedKafaJavaWrapperTest extends FlatSpec with Matchers {
  val props = new Properties()
  val testTopic = "test-topic"

  "A running server" should "return a list of topics" in {
    props.setProperty("bootstrap.servers", "localhost:10001")
    props.setProperty("delete.enable.topic", "true")
    props.setProperty("group.id", "test-client")
    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer")
    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty("clinet.id", "test-client")
    props.setProperty("key.serializer", "org.apache.kafka.common.serialization.LongSerializer")
    props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    EmbeddedKafaJavaWrapper.start(10001, 10002, props)

    try {
      implicit val admin = AdminClient.create(props)

      // create topic and confirm it exists
      createTopic(testTopic)
      val topics = listTopics()
      info(s"topics: $topics")
      topics should contain(testTopic)

      // now we should be able to send something to this topic
      // TODO create producer and send something

      // delete topic
      deleteTopic(testTopic)
      listTopics() shouldNot contain(testTopic)

      // recreate topic
      createTopic(testTopic)
      // listTopics() should contain(testTopic)

      // and finally consume from the topic and expect to get 0 entries
      // TODO create consumer and poll once
    } finally {
      EmbeddedKafaJavaWrapper.stop()
    }

  }

  def listTopics()(implicit admin: AdminClient) =
    admin.listTopics().names().get()

  def createTopic(topic: String)(implicit admin: AdminClient) =
    admin.createTopics(Seq(new NewTopic(topic, 1, 1)).asJava)

  def deleteTopic(topic: String)(implicit admin: AdminClient) =
    admin.deleteTopics(Seq("test-topic").asJava).all().get()

}
{code}

Btw, what happens to connected consumers when I delete a topic? 


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)