@Override
    public void commitAsync() {
        commitAsync(null);
    }
@Test
    public void testCommitAsyncWithFencedException() {
        consumer = newConsumer();
        completeCommitSyncApplicationEventSuccessfully();
        final Map<TopicPartition, OffsetAndMetadata> offsets = mockTopicPartitionOffset();
        MockCommitCallback callback = new MockCommitCallback();

        assertDoesNotThrow(() -> consumer.commitAsync(offsets, callback));

        final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor = ArgumentCaptor.forClass(AsyncCommitEvent.class);
        verify(applicationEventHandler).add(commitEventCaptor.capture());
        final AsyncCommitEvent commitEvent = commitEventCaptor.getValue();
        commitEvent.future().completeExceptionally(Errors.FENCED_INSTANCE_ID.exception());

        assertThrows(Errors.FENCED_INSTANCE_ID.exception().getClass(), () -> consumer.commitAsync());
    }