public Map<Consumer, List<EntryAndMetadata>> assign(final List<EntryAndMetadata> entryAndMetadataList,
                                                        final int numConsumers) {
        assert numConsumers >= 0;
        consumerToPermits.clear();
        final Map<Consumer, List<EntryAndMetadata>> consumerToEntries = new IdentityHashMap<>();

        Consumer consumer = getConsumer(numConsumers);
        if (consumer == null) {
            entryAndMetadataList.forEach(EntryAndMetadata::release);
            return consumerToEntries;
        }
        // The actual available permits might change, here we use the permits at the moment to assign entries
        int availablePermits = consumerToPermits.computeIfAbsent(consumer, Consumer::getAvailablePermits);
        int index = 0;
        for (; index < entryAndMetadataList.size(); index++) {
            final EntryAndMetadata entryAndMetadata = entryAndMetadataList.get(index);
            final MessageMetadata metadata = entryAndMetadata.getMetadata();

            // Select another consumer to ensure `consumer != null` and `availablePermits > 0`
            if (availablePermits <= 0) {
                consumerToPermits.put(consumer, availablePermits);
                consumer = getConsumer(numConsumers);
                if (consumer == null) {
                    break;
                }
                availablePermits = consumer.getAvailablePermits();
            }

            if (metadata == null || !metadata.hasUuid() || !metadata.hasChunkId() || !metadata.hasNumChunksFromMsg()) {
                consumerToEntries.computeIfAbsent(consumer, __ -> new ArrayList<>()).add(entryAndMetadata);
            } else {
                final Consumer consumerForUuid = getConsumerForUuid(metadata, consumer, availablePermits);
                if (consumerForUuid == null) {
                    unassignedMessageProcessor.accept(entryAndMetadata);
                    continue;
                }
                consumerToEntries.computeIfAbsent(consumerForUuid, __ -> new ArrayList<>()).add(entryAndMetadata);
            }
            availablePermits--;
        }

        for (; index < entryAndMetadataList.size(); index++) {
            unassignedMessageProcessor.accept(entryAndMetadataList.get(index));
        }

        return consumerToEntries;
    }
@Test
    public void testSingleConsumerMultiAssign() {
        // Only first 5 entries can be received because the number of permits is 5
        final Consumer consumer = new Consumer("A", 5);
        roundRobinConsumerSelector.addConsumers(consumer);

        Map<Consumer, List<EntryAndMetadata>> result = assignor.assign(entryAndMetadataList, 1);
        assertEquals(result.getOrDefault(consumer, Collections.emptyList()), entryAndMetadataList.subList(0, 5));
        // Since two chunked messages (A-1 and B-1) are both not received, these uuids have been cached
        assertEquals(assignor.getUuidToConsumer().keySet(), Sets.newHashSet("A-1", "B-1"));
        assertEquals(toString(replayQueue), Arrays.asList("0:5@A-1-2-3", "0:6@B-1-1-2"));

        result = assignor.assign(entryAndMetadataList.subList(5, 6), 1);
        assertEquals(result.getOrDefault(consumer, Collections.emptyList()), entryAndMetadataList.subList(5, 6));
        // A-1 is received so that uuid "A-1" has been removed from the cache
        assertEquals(assignor.getUuidToConsumer().keySet(), Sets.newHashSet("B-1"));

        result = assignor.assign(entryAndMetadataList.subList(6, 7), 1);
        assertEquals(result.getOrDefault(consumer, Collections.emptyList()), entryAndMetadataList.subList(6, 7));
        assertTrue(assignor.getUuidToConsumer().isEmpty());
    }