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

Code before:
        int maxNumTopicRecreationAttempts = 10;
        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, 1, (short) 1))).topicId(topic).get());
                } catch (Exception e) {
                    // ignore
                }
            }
            return topicIds;
        });

        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(topic, resp.topic());
                }
            } catch (Exception e) {
                // ignore
            }
        });
        var topicIds = recreateTopicFuture.join();
        producerFuture.join();
        assertEquals(maxNumTopicRecreationAttempts, topicIds.size());
        assertEquals(numRecords, numSuccess.intValue());
    }

    @ClusterTest
    public void testSendWithTopicReassignmentIsMidWay() throws Exception {
        var partition0 = new TopicPartition(topic, 0);

        try (var admin = cluster.admin();
             var producer = createProducer()
        ) {
            // Create topic with leader as 0 for the 1 partition.
            admin.createTopics(List.of(new NewTopic(topic, Map.of(0, List.of(0)))));
            assertLeader(partition0, 0);

            var topicDetails = topicMetadata();
            for (var i = 1; i <= numRecords; i++) {

Review comment (bug):
It's possible for get() to throw an exception like TopicNotExist, right? If that happens, the assertion on numRecords will fail.

Code after:
            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
    public void testSendWithTopicReassignmentIsMidWay() throws Exception {
        var partition0 = new TopicPartition(topic, 0);

        try (var admin = cluster.admin();
             var producer = createProducer()
        ) {
            // Create topic with leader as 0 for the 1 partition.
            admin.createTopics(List.of(new NewTopic(topic, Map.of(0, List.of(0)))));
            assertLeader(partition0, 0);

            var topicDetails = topicMetadata();
            for (var i = 1; i <= numRecords; i++) {
                var resp = producer.send(new ProducerRecord<>(topic, null, ("value" + i).getBytes())).get();
                assertEquals(topic, resp.topic());
            }

            var reassignment = Map.of(
                partition0, Optional.of(new NewPartitionReassignment(List.of(1)))
            );