@Override
      public void run() {
        if (endOfStreamPartitions.size() == numPartitions) {
          // shutdown scheduled executor as EOS is reached for all partitions
          executor.shutdownNow();
        } else {
          messageQueuesByPartition.forEach((partitionId, messages) -> {
            if (messages.isEmpty() && !endOfStreamPartitions.contains(partitionId)) {
              // if end of input is reached, send EOS for the partition and add the partition to the EOS partition set
              // to ensure we don't send EOS again for the partition
              producer.send(systemStream.getSystem(),
                  new OutgoingMessageEnvelope(systemStream, Integer.valueOf(partitionId), null, new EndOfStreamMessage(null)));
              endOfStreamPartitions.add(partitionId);
            } else if (!messages.isEmpty()) {
              final StreamMessageType e = messageQueuesByPartition.get(partitionId).removeFirst();
              final Object key = e instanceof KV ? ((KV) e).getKey() : null;
              final Object value = e instanceof KV ? ((KV) e).getValue() : e;
              if (value instanceof IncomingMessageEnvelope) {
                producer.send((IncomingMessageEnvelope) value);
              } else {
                producer.send(systemStream.getSystem(), new OutgoingMessageEnvelope(systemStream, Integer.valueOf(partitionId), key, value));
              }
            }
          });
        }
      }
Creates an in memory stream with {@link InMemorySystemFactory} and feeds its partition with stream of messages
@param partitionData key of the map represents partitionId and value represents messages in the partition
@param descriptor describes a stream to initialize with the in memory system
samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java