void onComplete(List<Map<TopicIdPartition, Acknowledgements>> acknowledgementsMapList) {
        final ArrayList<Throwable> exceptions = new ArrayList<>();
        acknowledgementsMapList.forEach(acknowledgementsMap -> acknowledgementsMap.forEach((partition, acknowledgements) -> {
            Exception exception = null;
            if (acknowledgements.getAcknowledgeErrorCode() != null) {
                exception = acknowledgements.getAcknowledgeErrorCode().exception();
            }
            Set<Long> offsets = acknowledgements.getAcknowledgementsTypeMap().keySet();
            Set<Long> offsetsCopy = Collections.unmodifiableSet(offsets);
            enteredCallback = true;
            try {
                acknowledgementCommitCallback.onComplete(Collections.singletonMap(partition, offsetsCopy), exception);
            } catch (Throwable e) {
                LOG.error("Exception thrown by acknowledgement commit callback", e);
                exceptions.add(e);
            } finally {
                enteredCallback = false;
            }
        }));
        if (!exceptions.isEmpty()) {
            throw ConsumerUtils.maybeWrapAsKafkaException(exceptions.get(0), "Exception thrown by acknowledgement commit callback");
        }
    }
@Test
    public void testNoException() throws Exception {
        Acknowledgements acknowledgements = Acknowledgements.empty();
        acknowledgements.add(0L, AcknowledgeType.ACCEPT);
        acknowledgements.add(1L, AcknowledgeType.REJECT);
        acknowledgementsMap.put(tip0, acknowledgements);

        acknowledgementCommitCallbackHandler.onComplete(Collections.singletonList(acknowledgementsMap));

        TestUtils.retryOnExceptionWithTimeout(() -> {
            assertNull(exceptionMap.get(tpo00));
            assertNull(exceptionMap.get(tpo01));
        });
    }