@Test
  public void shouldFailWhenNewProcessorJoinsWithSameIdAsExistingProcessor() throws InterruptedException {
    // Set up kafka topics.
    publishKafkaEvents(inputKafkaTopic, 0, NUM_KAFKA_EVENTS, PROCESSOR_IDS[0]);

    // Create StreamApplications.
    CountDownLatch kafkaEventsConsumedLatch = new CountDownLatch(NUM_KAFKA_EVENTS);
    CountDownLatch processedMessagesLatch1 = new CountDownLatch(1);
    CountDownLatch processedMessagesLatch2 = new CountDownLatch(1);

    ApplicationRunner appRunner1 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch1, null, kafkaEventsConsumedLatch,
        applicationConfig1), applicationConfig1);
    ApplicationRunner appRunner2 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch2, null, kafkaEventsConsumedLatch,
        applicationConfig2), applicationConfig2);

    // Run stream applications.
    executeRun(appRunner1, applicationConfig1);
    executeRun(appRunner2, applicationConfig2);

    // Wait for message processing to run in both the processors.
    processedMessagesLatch1.await();
    processedMessagesLatch2.await();

    // Create a stream app with same processor id as SP2 and run it. It should fail.
    publishKafkaEvents(inputKafkaTopic, NUM_KAFKA_EVENTS, 2 * NUM_KAFKA_EVENTS, PROCESSOR_IDS[2]);
    kafkaEventsConsumedLatch = new CountDownLatch(NUM_KAFKA_EVENTS);
    ApplicationRunner appRunner3 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, null, null, kafkaEventsConsumedLatch,
        applicationConfig2), applicationConfig2);
    // Fail when the duplicate processor joins.
    expectedException.expect(SamzaException.class);
    try {
      executeRun(appRunner3, applicationConfig2);
    } finally {
      appRunner1.kill();
      appRunner2.kill();

      appRunner1.waitForFinish();
      appRunner2.waitForFinish();
    }
  }
sspGrouper is set to AllSspToSingleTaskGrouperFactory (All ssps from input kafka topic are mapped to a single task per container).
AllSspToSingleTaskGrouperFactory should be used only with high-level consumers which do the partition management
by themselves. Using the factory with the consumers that do not do the partition management will result in
each processor/task consuming all the messages from all the partitions.
Run a stream application(streamApp1) consuming messages from input topic(effectively one container).

In the callback triggered by streamApp1 after processing a message, bring up an another stream application(streamApp2).

Assertions:
          A) JobModel generated before and after the addition of streamApp2 should not be equal.
          B) Second stream application(streamApp2) should join the group and process all the messages.
samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java