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 testMultiConsumerWithSmallPermits() {
        final Consumer consumerA = new Consumer("A", 3);
        final Consumer consumerB = new Consumer("B", 4);
        roundRobinConsumerSelector.addConsumers(consumerA, consumerB);

        // The original order:
        //   A-0, A-1-0-3, A-1-1-3, B-0, B-1-0-2, A-1-2-3, B-1-1-2
        // First consumerA received 3 entries and the available permits became 0. Then the consumer is switched to
        // consumerB and then B-0 and B-1-0-2 were received by it.
        // When consumerB tried to receive A-1-2-3, since the uuid "A-1" was already assigned to consumerA, consumerB
        // is not able to receive A-1-2-3.
        // However, since consumerA has no more permits, A-1-2-3 cannot be assigned to consumerA as well.
        // Therefore, A-1-2-3 was skipped and added to the replay queue.
        Map<Consumer, List<EntryAndMetadata>> result = assignor.assign(entryAndMetadataList, 2);
        assertEquals(toString(result.getOrDefault(consumerA, Collections.emptyList())),
                Arrays.asList("0:0@A-0", "0:1@A-1-0-3", "0:2@A-1-1-3"));
        assertEquals(toString(result.getOrDefault(consumerB, Collections.emptyList())),
                Arrays.asList("0:3@B-0", "0:4@B-1-0-2", "0:6@B-1-1-2"));
        assertEquals(toString(replayQueue), Collections.singletonList("0:5@A-1-2-3"));
        assertEquals(assignor.getUuidToConsumer().keySet(), Sets.newHashSet("A-1"));

        roundRobinConsumerSelector.clear();
        roundRobinConsumerSelector.addConsumers(consumerB, consumerA);
        assertSame(roundRobinConsumerSelector.peek(), consumerB);
        final List<EntryAndMetadata> leftEntries = new ArrayList<>(replayQueue);
        replayQueue.clear();
        result = assignor.assign(leftEntries, 2);
        // Since uuid "A-1" is still cached, A-1-2-3 will be dispatched to consumerA even if consumerB is the first
        // consumer returned by roundRobinConsumerSelector.get()
        assertEquals(toString(result.getOrDefault(consumerA, Collections.emptyList())),
                Collections.singletonList("0:5@A-1-2-3"));
        assertNull(result.get(consumerB));
        assertTrue(replayQueue.isEmpty());
        assertTrue(assignor.getUuidToConsumer().isEmpty());
    }