InMemorySystemConsumer implements SystemConsumer { @Override public Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll( Set<SystemStreamPartition> systemStreamPartitions, long timeout) throws InterruptedException { Map<SystemStreamPartition, String> sspOffsetPairToFetch = sspToOffset.entrySet() .stream() .filter(entry -> systemStreamPartitions.contains(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); Map<SystemStreamPartition, List<IncomingMessageEnvelope>> result = memoryManager.poll(sspOffsetPairToFetch); for (Map.Entry<SystemStreamPartition, List<IncomingMessageEnvelope>> sspToMessage : result.entrySet()) { sspToOffset.computeIfPresent(sspToMessage.getKey(), (ssp, offset) -> { int newOffset = Integer.parseInt(offset) + sspToMessage.getValue().size(); return String.valueOf(newOffset); }); } return result; } InMemorySystemConsumer(InMemoryManager manager); @Override void start(); @Override void stop(); @Override void register(SystemStreamPartition systemStreamPartition, String offset); @Override Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(
      Set<SystemStreamPartition> systemStreamPartitions, long timeout);  }
@Test public void testPoll() throws InterruptedException { this.inMemorySystemConsumer.register(SSP0, "1"); this.inMemorySystemConsumer.register(SSP1, "1"); IncomingMessageEnvelope ime01 = new IncomingMessageEnvelope(SSP0, "1", "key01", "message01"); IncomingMessageEnvelope ime02 = new IncomingMessageEnvelope(SSP0, "2", "key02", "message02"); Map<SystemStreamPartition, String> pollRequest = ImmutableMap.of(SSP0, "1"); when(this.inMemoryManager.poll(pollRequest)) .thenReturn(ImmutableMap.of(SSP0, ImmutableList.of())) .thenReturn(ImmutableMap.of(SSP0, ImmutableList.of(ime01, ime02))); pollRequest = ImmutableMap.of(SSP0, "3", SSP1, "1"); IncomingMessageEnvelope ime03 = new IncomingMessageEnvelope(SSP0, "3", "key03", "message03"); IncomingMessageEnvelope ime10 = new IncomingMessageEnvelope(SSP1, "1", "key10", "message10"); when(this.inMemoryManager.poll(pollRequest)).thenReturn( ImmutableMap.of(SSP0, ImmutableList.of(ime03), SSP1, ImmutableList.of(ime10))); assertEquals(ImmutableMap.of(SSP0, ImmutableList.of()), this.inMemorySystemConsumer.poll(ImmutableSet.of(SSP0), 1000)); assertEquals(ImmutableMap.of(SSP0, ImmutableList.of(ime01, ime02)), this.inMemorySystemConsumer.poll(ImmutableSet.of(SSP0), 1000)); assertEquals(ImmutableMap.of(SSP0, ImmutableList.of(ime03), SSP1, ImmutableList.of(ime10)), this.inMemorySystemConsumer.poll(ImmutableSet.of(SSP0, SSP1), 1000)); }