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 testMultiplePartitions() throws Exception {
        Acknowledgements acknowledgements = Acknowledgements.empty();
        acknowledgements.add(0L, AcknowledgeType.ACCEPT);
        acknowledgements.add(1L, AcknowledgeType.REJECT);
        acknowledgements.setAcknowledgeErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED);
        acknowledgementsMap.put(tip0, acknowledgements);

        Acknowledgements acknowledgements1 = Acknowledgements.empty();
        acknowledgements1.add(0L, AcknowledgeType.RELEASE);
        acknowledgements1.setAcknowledgeErrorCode(Errors.INVALID_RECORD_STATE);
        acknowledgementsMap.put(tip1, acknowledgements1);

        Map<TopicIdPartition, Acknowledgements> acknowledgementsMap2 = new HashMap<>();
        Acknowledgements acknowledgements2 = Acknowledgements.empty();
        acknowledgements2.add(0L, AcknowledgeType.ACCEPT);
        acknowledgementsMap2.put(tip2, acknowledgements2);

        List<Map<TopicIdPartition, Acknowledgements>> acknowledgementsMapList = new LinkedList<>();
        acknowledgementsMapList.add(acknowledgementsMap);
        acknowledgementsMapList.add(acknowledgementsMap2);
        acknowledgementCommitCallbackHandler.onComplete(acknowledgementsMapList);
        TestUtils.retryOnExceptionWithTimeout(() -> {
            assertInstanceOf(TopicAuthorizationException.class, exceptionMap.get(tpo00));
            assertInstanceOf(TopicAuthorizationException.class, exceptionMap.get(tpo01));
            assertInstanceOf(InvalidRecordStateException.class, exceptionMap.get(tpo10));
            assertNull(exceptionMap.get(tpo20));
        });
    }