InMemoryManager { Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(Map<SystemStreamPartition, String> sspsToOffsets) { return sspsToOffsets.entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> poll(entry.getKey(), entry.getValue()))); } InMemoryManager();   }
@Test public void testPoll() { this.inMemoryManager.initializeStream(new StreamSpec(STREAM0, STREAM0, SYSTEM, 1)); this.inMemoryManager.initializeStream(new StreamSpec(STREAM1, STREAM1, SYSTEM, 1)); this.inMemoryManager.initializeStream(new StreamSpec("otherStream", "otherStream", SYSTEM, 1)); SystemStreamPartition ssp0 = new SystemStreamPartition(SYSTEM, STREAM0, new Partition(0)); assertEquals(ImmutableMap.of(ssp0, ImmutableList.of()), this.inMemoryManager.poll(Collections.singletonMap(ssp0, "0"))); this.inMemoryManager.put(ssp0, "key00", "message00"); Map<SystemStreamPartition, List<IncomingMessageEnvelope>> polledMessages = this.inMemoryManager.poll(Collections.singletonMap(ssp0, "0")); assertEquals(1, polledMessages.get(ssp0).size()); assertIncomingMessageEnvelope("key00", "message00", "0", ssp0, polledMessages.get(ssp0).get(0)); this.inMemoryManager.put(ssp0, "key01", "message01"); polledMessages = this.inMemoryManager.poll(ImmutableMap.of(ssp0, "0")); assertEquals(2, polledMessages.get(ssp0).size()); assertIncomingMessageEnvelope("key00", "message00", "0", ssp0, polledMessages.get(ssp0).get(0)); assertIncomingMessageEnvelope("key01", "message01", "1", ssp0, polledMessages.get(ssp0).get(1)); polledMessages = this.inMemoryManager.poll(ImmutableMap.of(ssp0, "1")); assertEquals(1, polledMessages.get(ssp0).size()); assertIncomingMessageEnvelope("key01", "message01", "1", ssp0, polledMessages.get(ssp0).get(0)); SystemStreamPartition ssp1 = new SystemStreamPartition(SYSTEM, STREAM1, new Partition(0)); this.inMemoryManager.put(ssp1, "key10", "message10"); polledMessages = this.inMemoryManager.poll(ImmutableMap.of(ssp0, "1", ssp1, "0")); assertEquals(1, polledMessages.get(ssp0).size()); assertIncomingMessageEnvelope("key01", "message01", "1", ssp0, polledMessages.get(ssp0).get(0)); assertEquals(1, polledMessages.get(ssp1).size()); assertIncomingMessageEnvelope("key10", "message10", "0", ssp1, polledMessages.get(ssp1).get(0)); }