OmniaGM commented on code in PR #19964:
URL: https://github.com/apache/kafka/pull/19964#discussion_r2160223873
##########
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)
Review Comment:
I changed the test a bit here now there is a counter for success records
that get updated in the callback and I verify this at the end of the test
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
