Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerSendWhileDeletionTest.java (Java)

Code before:
            }

            // Start topic deletion
            admin.deleteTopics(List.of(topic)).all().get();

            // Verify that the topic is deleted when no metadata request comes in
            verifyTopicDeletion();
            cluster.createTopic(topic, 1, (short) 1);
            assertNotEquals(topicId, topicMetadata().topicId());

            // Producer should be able to send messages even after topic gets recreated
            var recordMetadata = producer.send(new ProducerRecord<>(topic, null, "value".getBytes(StandardCharsets.UTF_8))).get();
            assertEquals(topic, recordMetadata.topic());
            assertEquals(0, recordMetadata.offset());
        }
    }

    @Timeout(90)
    @ClusterTest
    public void testSendWhileTopicGetRecreated() {
        int maxNumRecreatTopicAttempts = 10;
        List<Uuid> topicIds = new CopyOnWriteArrayList<>();
        var recreateTopicFuture = CompletableFuture.runAsync(() -> {
            for (int i = 1; i <= maxNumRecreatTopicAttempts; i++) {
                Uuid topicId = recreateTopic();
                if (topicId != Uuid.ZERO_UUID) {
                    topicIds.add(topicId);
                }
            }
        });

        AtomicInteger numSuccess = new AtomicInteger(0);
        var producerFuture = CompletableFuture.runAsync(() -> {
            try (var producer = cluster.producer()) {
                for (int i = 1; i <= numRecords; i++) {
                    var resp = producer.send(new ProducerRecord<>(topic, null, ("value" + i).getBytes()),
                            (metadata, exception) -> {
                                if (metadata != null) {
                                    numSuccess.incrementAndGet();
                                }
                            }).get();
                    assertEquals(resp.topic(), topic);
                }
            } catch (Exception e) {
                // ignore
            }
        });
        recreateTopicFuture.join();
        producerFuture.join();
        assertTrue(Math.abs(maxNumRecreatTopicAttempts - topicIds.size()) <= 5);
        assertEquals(10, numSuccess.intValue());

Review comment (suggestion):
Should we compare using `equals()`:

```suggestion
                if (!topicId.equals(Uuid.ZERO_UUID)) {
```

Code after:
            admin.deleteTopics(List.of(topic)).all().get();

            // Verify that the topic is deleted when no metadata request comes in
            verifyTopicDeletion();
            cluster.createTopic(topic, 1, (short) 1);
            assertNotEquals(topicId, topicMetadata().topicId());

            // Producer should be able to send messages even after topic gets recreated
            var recordMetadata = producer.send(new ProducerRecord<>(topic, null, "value".getBytes(StandardCharsets.UTF_8))).get();
            assertEquals(topic, recordMetadata.topic());
            assertEquals(0, recordMetadata.offset());
        }
    }

    @ClusterTest
    public void testSendWhileTopicGetRecreated() {
        int maxNumTopicRecreationAttempts = 5;
        var recreateTopicFuture = CompletableFuture.supplyAsync(() -> {
            var topicIds = new HashSet<Uuid>();
            while (topicIds.size() < maxNumTopicRecreationAttempts) {
                try (var admin = cluster.admin()) {
                    if (admin.listTopics().names().get().contains(topic)) {
                        admin.deleteTopics(List.of(topic)).all().get();
                    }
                    topicIds.add(admin.createTopics(List.of(new NewTopic(topic, 2, (short) 1))).topicId(topic).get());
                } catch (Exception e) {
                    // ignore
                }
            }
            return topicIds;
        });

        AtomicInteger numAcks = new AtomicInteger(0);
        var producerFuture = CompletableFuture.runAsync(() -> {
            try (var producer = createProducer()) {
                for (int i = 1; i <= numRecords; i++) {
                    producer.send(new ProducerRecord<>(topic, null, ("value" + i).getBytes()),
                            (metadata, exception) -> {
                                numAcks.incrementAndGet();
                            });
                }
                producer.flush();
            }
        });
        var topicIds = recreateTopicFuture.join();
        producerFuture.join();
        assertEquals(maxNumTopicRecreationAttempts, topicIds.size());
        assertEquals(numRecords, numAcks.intValue());
    }

    @ClusterTest