OmniaGM commented on code in PR #19964:
URL: https://github.com/apache/kafka/pull/19964#discussion_r2160223692
##########
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 = {
Review Comment:
made it private as moving the test to the new java test
##########
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 =>
Review Comment:
This was left from the recreation logic of the topic I just handled this in
the recreate
--
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
