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

Code before:
            otherConsumer.assign(List.of(tp));
            consumeAndVerifyRecords(otherConsumer, tp, 1, 5, 0, startingTimestamp);
        }
    }

    @ClusterTest
    public void testClearBrokerBeforeConsumerCloses() throws InterruptedException {
        // This is testing when closing the consumer but commit request has already been sent.
        // During the closing, the consumer won't find the coordinator anymore.

        // Create offsets topic to ensure coordinator is available during close
        cluster.createTopic(Topic.GROUP_METADATA_TOPIC_NAME, Integer.parseInt(OFFSETS_TOPIC_PARTITIONS), Short.parseShort(OFFSETS_TOPIC_REPLICATION));

        try (Producer<byte[], byte[]> producer = cluster.producer(Map.of(ProducerConfig.ACKS_CONFIG, "all"));
             var consumer = createConsumer(GroupProtocol.CONSUMER, false)
        ) {
            sendRecords(producer, tp, 3, System.currentTimeMillis());
            sendRecords(producer, tp1, 3, System.currentTimeMillis());
            consumer.assign(List.of(tp, tp1));

            // Try without looking up the coordinator first
            var cb = new CountConsumerCommitCallback();

            // Close the coordinator before committing because otherwise the commit will fail to find the coordinator.
            Set<Integer> brokerIds = cluster.brokerIds();
            brokerIds.forEach(cluster::shutdownBroker);

            consumer.commitAsync(Map.of(tp, new OffsetAndMetadata(1L)), cb);
            consumer.commitAsync(Map.of(tp1, new OffsetAndMetadata(1L)), cb);

            consumer.close();
            assertTrue(cb.lastError.get() instanceof CommitFailedException);
            assertEquals(0, cb.successCount);
        }
    }

    // TODO: This only works in the new consumer, but should be fixed for the old consumer as well
    @ClusterTest
    public void testCommitAsyncCompletedBeforeConsumerCloses() throws InterruptedException {
        // This is testing the contract that asynchronous offset commit are completed before the consumer
        // is closed, even when no commit sync is performed as part of the close (due to auto-commit
        // disabled, or simply because there are no consumed offsets).

        // Create offsets topic to ensure coordinator is available during close
        cluster.createTopic(Topic.GROUP_METADATA_TOPIC_NAME, Integer.parseInt(OFFSETS_TOPIC_PARTITIONS), Short.parseShort(OFFSETS_TOPIC_REPLICATION));

        try (Producer<byte[], byte[]> producer = cluster.producer(Map.of(ProducerConfig.ACKS_CONFIG, "all"));
             var consumer = createConsumer(GroupProtocol.CONSUMER, false)
        ) {
            sendRecords(producer, tp, 3, System.currentTimeMillis());
            sendRecords(producer, tp1, 3, System.currentTimeMillis());

Review comment (suggestion):
```suggestion
            cluster.brokerIds().forEach(cluster::shutdownBroker);
```

Code after:
            // another consumer in the same group should get the same position
            otherConsumer.assign(List.of(tp));
            consumeAndVerifyRecords(otherConsumer, tp, 1, 5, 0, startingTimestamp);
        }
    }

    /**
     * This is testing when closing the consumer but commit request has already been sent.
     * During the closing, the consumer won't find the coordinator anymore.
     */
    @ClusterTest
    public void testCommitAsyncFailsWhenCoordinatorUnavailableDuringClose() throws InterruptedException {
        try (Producer<byte[], byte[]> producer = cluster.producer();
             var consumer = createConsumer(GroupProtocol.CONSUMER, false)
        ) {
            sendRecords(producer, tp, 3, System.currentTimeMillis());
            consumer.assign(List.of(tp));

            var callback = new CountConsumerCommitCallback();

            // Close the coordinator before committing because otherwise the commit will fail to find the coordinator.
            cluster.brokerIds().forEach(cluster::shutdownBroker);

            TestUtils.waitForCondition(() -> cluster.aliveBrokers().isEmpty(), "All brokers should be shut down");

            consumer.poll(Duration.ofMillis(500));
            consumer.commitAsync(Map.of(tp, new OffsetAndMetadata(1L)), callback);

            long startTime = System.currentTimeMillis();
            consumer.close(CloseOptions.timeout(Duration.ofMillis(500)));
            long closeDuration = System.currentTimeMillis() - startTime;

            assertTrue(closeDuration < 1000, "The closing process for the consumer was too long: " + closeDuration + " ms");
            assertTrue(callback.lastError.isPresent());
            assertEquals(CommitFailedException.class, callback.lastError.get().getClass());
            assertEquals("Failed to commit offsets: Coordinator unknown and consumer is closing", callback.lastError.get().getMessage());
            assertEquals(1, callback.exceptionCount);
        }
    }

    // TODO: This only works in the new consumer, but should be fixed for the old consumer as well
    @ClusterTest
    public void testCommitAsyncCompletedBeforeConsumerCloses() throws InterruptedException {
        // This is testing the contract that asynchronous offset commit are completed before the consumer
        // is closed, even when no commit sync is performed as part of the close (due to auto-commit
        // disabled, or simply because there are no consumed offsets).

        // Create offsets topic to ensure coordinator is available during close
        cluster.createTopic(Topic.GROUP_METADATA_TOPIC_NAME, Integer.parseInt(OFFSETS_TOPIC_PARTITIONS), Short.parseShort(OFFSETS_TOPIC_REPLICATION));

        try (Producer<byte[], byte[]> producer = cluster.producer(Map.of(ProducerConfig.ACKS_CONFIG, "all"));