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 as it deletes the metadata first however
`deleteTopic` without the admin raw waits for delete 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
