@Test
  public void shouldStopNewProcessorsJoiningGroupWhenNumContainersIsGreaterThanNumTasks() throws InterruptedException {
    // Set up kafka topics.
    publishKafkaEvents(inputSinglePartitionKafkaTopic, 0, NUM_KAFKA_EVENTS * 2, PROCESSOR_IDS[0]);

    // Configuration, verification variables
    MapConfig testConfig = new MapConfig(ImmutableMap.of(JobConfig.SSP_GROUPER_FACTORY,
        "org.apache.samza.container.grouper.stream.GroupBySystemStreamPartitionFactory", JobConfig.JOB_DEBOUNCE_TIME_MS, "10",
            ClusterManagerConfig.JOB_HOST_AFFINITY_ENABLED, "true"));
    // Declared as final array to update it from streamApplication callback(Variable should be declared final to access in lambda block).
    final JobModel[] previousJobModel = new JobModel[1];
    final String[] previousJobModelVersion = new String[1];
    AtomicBoolean hasSecondProcessorJoined = new AtomicBoolean(false);
    final CountDownLatch secondProcessorRegistered = new CountDownLatch(1);

    zkUtils.subscribeToProcessorChange((parentPath, currentChilds) -> {
      // When appRunner2 with id: PROCESSOR_IDS[1] is registered, run processing message in appRunner1.
      if (currentChilds.contains(PROCESSOR_IDS[1])) {
        secondProcessorRegistered.countDown();
      }
    });

    // Set up stream app appRunner2.
    CountDownLatch processedMessagesLatch = new CountDownLatch(NUM_KAFKA_EVENTS);
    Config localTestConfig2 = new MapConfig(applicationConfig2, testConfig);
    ApplicationRunner appRunner2 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputSinglePartitionKafkaTopic), outputSinglePartitionKafkaTopic, processedMessagesLatch,
        null, null, localTestConfig2), localTestConfig2);

    // Callback handler for appRunner1.
    TestStreamApplication.StreamApplicationCallback callback = m -> {
      if (hasSecondProcessorJoined.compareAndSet(false, true)) {
        previousJobModelVersion[0] = zkUtils.getJobModelVersion();
        previousJobModel[0] = JobModelUtil.readJobModel(previousJobModelVersion[0], zkMetadataStore);
        executeRun(appRunner2, localTestConfig2);
        try {
          // Wait for appRunner2 to register with zookeeper.
          secondProcessorRegistered.await();
        } catch (InterruptedException e) {
        }
      }
    };

    CountDownLatch kafkaEventsConsumedLatch = new CountDownLatch(NUM_KAFKA_EVENTS * 2);

    // Set up stream app appRunner1.
    Config localTestConfig1 = new MapConfig(applicationConfig1, testConfig);
    ApplicationRunner appRunner1 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputSinglePartitionKafkaTopic), outputSinglePartitionKafkaTopic, null,
        callback, kafkaEventsConsumedLatch, localTestConfig1), localTestConfig1);
    executeRun(appRunner1, localTestConfig1);

    kafkaEventsConsumedLatch.await();

    String currentJobModelVersion = zkUtils.getJobModelVersion();
    JobModel updatedJobModel = JobModelUtil.readJobModel(currentJobModelVersion, zkMetadataStore);

    // Job model before and after the addition of second stream processor should be the same.
    assertEquals(previousJobModel[0], updatedJobModel);
    assertEquals(new MapConfig(), updatedJobModel.getConfig());
    assertEquals(NUM_KAFKA_EVENTS, processedMessagesLatch.getCount());
    appRunner2.kill();
    appRunner2.waitForFinish();
    assertEquals(appRunner2.status(), ApplicationStatus.UnsuccessfulFinish);
    appRunner1.kill();
    appRunner1.waitForFinish();
    assertEquals(appRunner1.status(), ApplicationStatus.SuccessfulFinish);
  }
sspGrouper is set to GroupBySystemStreamPartitionFactory.
Run a stream application(appRunner1) consuming messages from input topic(effectively one container).

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

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