@Override
  public Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(Set<SystemStreamPartition> set, long l) throws InterruptedException {
    if (!done) {
      Map<SystemStreamPartition, List<IncomingMessageEnvelope>> envelopeMap = new HashMap<>();
      final AtomicInteger offset = new AtomicInteger(0);
      set.forEach(ssp -> {
        List<IncomingMessageEnvelope> envelopes = Arrays.stream(getArrayObjects(ssp.getSystemStream().getStream(), config))
            .map(object -> new IncomingMessageEnvelope(ssp, String.valueOf(offset.incrementAndGet()), null, object)).collect(Collectors.toList());
        envelopes.add(IncomingMessageEnvelope.buildEndOfStreamEnvelope(ssp));
        envelopeMap.put(ssp, envelopes);
      });
      done = true;
      return envelopeMap;
    } else {
      return Collections.emptyMap();
    }

  }
A simple implementation of array system consumer
samza-test/src/test/java/org/apache/samza/test/util/ArraySystemConsumer.java