OmniaGM commented on code in PR #19964:
URL: https://github.com/apache/kafka/pull/19964#discussion_r2160224506
##########
core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala:
##########
@@ -81,6 +86,59 @@ class ProducerSendWhileDeletionTest extends
IntegrationTestHarness {
assertEquals(topic, producer.send(new ProducerRecord(topic, null,
"value".getBytes(StandardCharsets.UTF_8))).get.topic())
}
+ @Test
+ def testSendWhileTopicGetRecreated(): Unit = {
+ val numRecords = 10
+ val topic = "topic"
+ val admin = createAdminClient()
+ val producer = createProducer()
+
+ try {
+ val fs = CompletableFuture.runAsync(() => {
+ for (_ <- 1 to 20) {
+ recreateTopic(admin, topic)
+ }
+ })
+ val producerFutures = new util.ArrayList[CompletableFuture[Void]]
+ for (_ <- 0 until numRecords) {
+ producerFutures.add(CompletableFuture.runAsync(() => {
+ for (i <- 0 until numRecords) {
+ producer.send(new ProducerRecord(topic, null, ("value" +
i).getBytes(StandardCharsets.UTF_8)), new Callback() {
+ override def onCompletion(metadata: RecordMetadata, exception:
Exception): Unit = {
+ assertNotNull(metadata)
+ assertNotEquals(metadata.offset, -1L)
+ assertNotEquals(metadata.timestamp, RecordBatch.NO_TIMESTAMP)
+ }
+ }).get()
+ }
+ }))
+ }
+ fs.join()
+ producerFutures.forEach(_.join)
+ } catch {
+ case e: TopicExistsException =>
+ admin.deleteTopics(util.List.of(topic)).all().get()
+ } finally {
+ admin.close()
+ producer.close()
+ }
+ }
+
+ def recreateTopic(admin: Admin, topic: String): Unit = {
+ waitForCondition(() => {
+ val topicId = if (admin.listTopics().names().get().contains(topic)) {
+ topicMetadata(admin, topic).topicId()
+ } else Uuid.ZERO_UUID
+ // don't wait for the physical delete
+ deleteTopicWithAdminRaw(admin, topic)
Review Comment:
Delete topic usually faster at deleting the metadata first however
`deleteTopic` without the admin raw waits for deleting the hard partitions
which might take longer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
