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

Code before:
            consumer.assign(List.of(tp));

            assertEquals(0L, consumer.position(tp), "position() on a partition that we are subscribed to should reset the offset");
            consumer.commitSync();
            assertEquals(0L, consumer.committed(Set.of(tp)).get(tp).offset());
            consumeAndVerifyRecords(consumer, tp, 5, 0, 0, startingTimestamp);
            assertEquals(5L, consumer.position(tp), "After consuming 5 records, position should be 5");
            consumer.commitSync();
            assertEquals(5L, consumer.committed(Set.of(tp)).get(tp).offset(), "Committed offset should be returned");

            startingTimestamp = System.currentTimeMillis();
            sendRecords(producer, tp, 1, startingTimestamp);

            // 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(Map.of(ProducerConfig.ACKS_CONFIG, "all"));
             var consumer = createConsumer(GroupProtocol.CONSUMER, false)
        ) {
            sendRecords(producer, tp, 3, System.currentTimeMillis());
            consumer.assign(List.of(tp));

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

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

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

            consumer.close(CloseOptions.timeout(Duration.ofMillis(500)));

            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(2, callback.exceptionCount);
        }
    }

    // TODO: This only works in the new consumer, but should be fixed for the old consumer as well
    @ClusterTest

Review comment (suggestion):
The default of acks is already `all` so we can remove it.

Code after:

            consumer.assign(List.of(tp));

            assertEquals(0L, consumer.position(tp), "position() on a partition that we are subscribed to should reset the offset");
            consumer.commitSync();
            assertEquals(0L, consumer.committed(Set.of(tp)).get(tp).offset());
            consumeAndVerifyRecords(consumer, tp, 5, 0, 0, startingTimestamp);
            assertEquals(5L, consumer.position(tp), "After consuming 5 records, position should be 5");
            consumer.commitSync();
            assertEquals(5L, consumer.committed(Set.of(tp)).get(tp).offset(), "Committed offset should be returned");

            startingTimestamp = System.currentTimeMillis();
            sendRecords(producer, tp, 1, startingTimestamp);

            // 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);