@Test
  public void testStreamProcessor() {
    final String testSystem = "test-system";
    final String inputTopic = "numbers";
    final String outputTopic = "output";
    final int messageCount = 20;

    final Config configs = new MapConfig(createConfigs("1", testSystem, inputTopic, outputTopic, messageCount));
    // Note: createTopics needs to be called before creating a StreamProcessor. Otherwise it fails with a
    // TopicExistsException since StreamProcessor auto-creates them.
    createTopics(inputTopic, outputTopic);
    final TestStubs stubs = new TestStubs(configs, IdentityStreamTask::new, bootstrapServers());

    produceMessages(stubs.producer, inputTopic, messageCount);
    run(stubs.processor, stubs.shutdownLatch);
    verifyNumMessages(stubs.consumer, outputTopic, messageCount);
  }
Testing a basic identity stream task - reads data from a topic and writes it to another topic
(without any modifications)

<p>
The standalone version in this test uses KafkaSystemFactory and it uses a SingleContainerGrouperFactory. Hence,
no matter how many tasks are present, it will always be run in a single processor instance. This simplifies testing
samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java