public Map<Integer, List<ProducerBatch>> drain(MetadataSnapshot metadataSnapshot, Set<Node> nodes, int maxSize, long now) {
        if (nodes.isEmpty())
            return Collections.emptyMap();

        Map<Integer, List<ProducerBatch>> batches = new HashMap<>();
        for (Node node : nodes) {
            List<ProducerBatch> ready = drainBatchesForOneNode(metadataSnapshot, node, maxSize, now);
            batches.put(node.id(), ready);
        }
        return batches;
    }
@Test
    public void testDrainWithANodeThatDoesntHostAnyPartitions() {
        int batchSize = 10;
        int lingerMs = 10;
        long totalSize = 10 * 1024;
        RecordAccumulator accum = createTestRecordAccumulator(batchSize, totalSize, Compression.NONE, lingerMs);

        // Create cluster metadata, node2 doesn't host any partitions.
        PartitionMetadata part1Metadata = new PartitionMetadata(Errors.NONE, tp1, Optional.of(node1.id()), Optional.empty(), null, null, null);
        MetadataSnapshot metadataCache = new MetadataSnapshot(null, nodes, Collections.singletonList(part1Metadata), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());

        // Drain for node2, it should return 0 batches,
        Map<Integer, List<ProducerBatch>> batches = accum.drain(metadataCache,
            new HashSet<>(Collections.singletonList(node2)), 999999 /* maxSize */, time.milliseconds());
        assertTrue(batches.get(node2.id()).isEmpty());
    }