private <StreamMessageType> void initializeInMemoryInputStream(InMemoryInputDescriptor<?> descriptor,
      Map<Integer, Iterable<StreamMessageType>> partitionData, long delayInMillis) {
    String systemName = descriptor.getSystemName();
    String streamName = (String) descriptor.getPhysicalName().orElse(descriptor.getStreamId());
    if (this.app instanceof LegacyTaskApplication) {
      // task.inputs is generated using descriptors for Task/StreamApplication, but needs to be generated here
      // for legacy applications that only specify task.class.
      if (configs.containsKey(TaskConfig.INPUT_STREAMS)) {
        configs.put(TaskConfig.INPUT_STREAMS,
            configs.get(TaskConfig.INPUT_STREAMS).concat("," + systemName + "." + streamName));
      } else {
        configs.put(TaskConfig.INPUT_STREAMS, systemName + "." + streamName);
      }
    }
    InMemorySystemDescriptor imsd = (InMemorySystemDescriptor) descriptor.getSystemDescriptor();
    imsd.withInMemoryScope(this.inMemoryScope);
    addConfig(descriptor.toConfig());
    addConfig(descriptor.getSystemDescriptor().toConfig());
    addSerdeConfigs(descriptor);
    StreamSpec spec = new StreamSpec(descriptor.getStreamId(), streamName, systemName, partitionData.size());
    SystemFactory factory = new InMemorySystemFactory();
    Config config = new MapConfig(descriptor.toConfig(), descriptor.getSystemDescriptor().toConfig());
    factory.getAdmin(systemName, config).createStream(spec);
    InMemorySystemProducer producer = (InMemorySystemProducer) factory.getProducer(systemName, config, null);
    SystemStream sysStream = new SystemStream(systemName, streamName);
    if (delayInMillis > 0) {
      delayedInitialization(partitionData, producer, sysStream, delayInMillis);
    } else {
      partitionData.forEach((partitionId, partition) -> {
        partition.forEach(e -> {
          Object key = e instanceof KV ? ((KV) e).getKey() : null;
          Object value = e instanceof KV ? ((KV) e).getValue() : e;
          if (value instanceof IncomingMessageEnvelope) {
            producer.send((IncomingMessageEnvelope) value);
          } else {
            producer.send(systemName, new OutgoingMessageEnvelope(sysStream, Integer.valueOf(partitionId), key, value));
          }
        });
        producer.send(systemName, new OutgoingMessageEnvelope(sysStream, Integer.valueOf(partitionId), null,
            new EndOfStreamMessage(null)));
      });
    }
  }
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