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

Code before:
                    topicIds.add(admin.createTopics(List.of(new NewTopic(topic, 1, (short) 1))).topicId(topic).get());
                } catch (Exception e) {
                    // ignore
                }
            }
            return topicIds;
        });

        Map<String, Object> producerConfig = Map.of(
                ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName(),
                "mock.interceptor.append", "",
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, MockSerializer.class.getName(),
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MockSerializer.class.getName());

        var producerFuture = CompletableFuture.runAsync(() -> {
            try (var producer = cluster.producer(producerConfig)) {
                for (int i = 1; i <= numRecords; i++) {
                    producer.send(new ProducerRecord<>(topic, String.valueOf(i), ("value" + i)));
                }
            }
        });
        var topicIds = recreateTopicFuture.join();
        producerFuture.join();
        assertEquals(maxNumTopicRecreationAttempts, topicIds.size());
        TestUtils.waitForCondition(() ->
                        numRecords == MockProducerInterceptor.ON_SUCCESS_COUNT.intValue(),
                "didn't produce all records");
    }

    @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)))
            );
            // Change replica assignment from 0 to 1. Leadership moves to 1.
            admin.alterPartitionReassignments(reassignment).all().get();

Review comment (bug):
Hmm, this still doesn't seem quite right. We are only sending numRecords. If some of the records fail to send, ON_SUCCESS_COUNT is never going to be numRecords. 

To me, the goal of this test is to make sure that the callback for each send is called, with or w/o an exception. Perhaps we could do sth like the following?

```
           for (int i = 1; i <= numRecords; i++) {
                    var resp = producer.send(new ProducerRecord<>(topic, null, ("value" + i).getBytes()),
                            (metadata, exception) -> {
                                    numAcks.incrementAndGet();
                                }
                            });
             }
             producer.flush();

          ....

assertEquals(numRecords, numAcks.intValue());
```

Code after:
            }
            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)))
            );
            // Change replica assignment from 0 to 1. Leadership moves to 1.
            admin.alterPartitionReassignments(reassignment).all().get();

            assertLeader(partition0, 1);
            assertEquals(topicDetails.topicId(), topicMetadata().topicId());

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