Add test testCommitAsync_NullCallback as testCommitAsyncWithNullCallback

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerUnitTest.java

@@ -75,6 +75,7 @@
 import org.apache.kafka.common.utils.Timer;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mockito;
 
@@ -219,6 +220,25 @@ public void testCommitSyncLeaderEpochUpdate() {
         verify(metadata).updateLastSeenEpochIfNewer(t1, 1);
     }
 
+    @Test
+    public void testCommitAsyncWithNullCallback() {
+        consumer = setup();
+        final TopicPartition t0 = new TopicPartition("t0", 2);
+        final TopicPartition t1 = new TopicPartition("t0", 3);
+        HashMap<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+        offsets.put(t0, new OffsetAndMetadata(10L));
+        offsets.put(t1, new OffsetAndMetadata(20L));
+
+        consumer.commitAsync(offsets, null);
+
+        final ArgumentCaptor<CommitApplicationEvent> commitEventCaptor = ArgumentCaptor.forClass(CommitApplicationEvent.class);
+        verify(applicationEventHandler).add(commitEventCaptor.capture());
+        final CommitApplicationEvent commitEvent = commitEventCaptor.getValue();
+        assertEquals(offsets, commitEvent.offsets());
+        assertDoesNotThrow(() -> commitEvent.future().complete(null));
+        assertDoesNotThrow(() -> consumer.commitAsync(offsets, null));
+    }
+
     @Test
     public void testCommitAsyncLeaderEpochUpdate() {
         consumer = setup();