@Test
  public void testSSPGrouperProxyUsed() {
    addStreamMetadataCacheMetadata(this.streamMetadataCache,
        ImmutableMap.of(SYSTEM_STREAM0, buildSystemStreamMetadata(4)));
    Map<TaskName, Integer> changelogPartitionMapping = changelogPartitionMapping(2);
    Config config = config(ImmutableList.of(SYSTEM_STREAM0),
        ImmutableMap.of(JobConfig.SSP_GROUPER_FACTORY, Partition0SeparateFactory.class.getName(),
            // need this to trigger SSPGrouperProxy logic
            String.format(StorageConfig.FACTORY, "myStore"), "MyCustomStore"));
    // custom SSP grouper expects a certain processor locality for another test, so add the locality here too
    when(this.grouperMetadata.getProcessorLocality()).thenReturn(
        ImmutableMap.of("0", mock(LocationId.class), "1", mock(LocationId.class)));
    /*
     * Even though the custom grouper factory would normally send the additional SSPs to task 1, the SSP grouper proxy
     * should give task 0 some of the SSPs.
     */
    when(this.grouperMetadata.getPreviousTaskToSSPAssignment()).thenReturn(
        ImmutableMap.of(taskName(0), ImmutableList.of(new SystemStreamPartition(SYSTEM_STREAM0, new Partition(0))),
            taskName(1), ImmutableList.of(new SystemStreamPartition(SYSTEM_STREAM0, new Partition(1)))));
    Set<SystemStreamPartition> sspsForTask0 =
        ImmutableSet.of(new SystemStreamPartition(SYSTEM_STREAM0, new Partition(0)),
            new SystemStreamPartition(SYSTEM_STREAM0, new Partition(2)));
    Set<SystemStreamPartition> sspsForTask1 =
        ImmutableSet.of(new SystemStreamPartition(SYSTEM_STREAM0, new Partition(1)),
            new SystemStreamPartition(SYSTEM_STREAM0, new Partition(3)));
    Map<String, ContainerModel> containerModels = ImmutableMap.of("0", new ContainerModel("0",
            ImmutableMap.of(taskName(0), new TaskModel(taskName(0), sspsForTask0, new Partition(0)))), "1",
        new ContainerModel("1",
            ImmutableMap.of(taskName(1), new TaskModel(taskName(1), sspsForTask1, new Partition(1)))));
    JobModel expected = new JobModel(config, containerModels);
    JobModel actual =
        JobModelCalculator.INSTANCE.calculateJobModel(config, changelogPartitionMapping, this.streamMetadataCache,
            this.grouperMetadata);
    assertEquals(expected, actual);
  }
Unfortunately, we still need powermock for testing the regex topic flow, since that flow checks for a specific
rewriter class, and setting up the support for that specific class is too cumbersome.
Many of the tests do not need powermock.
samza-core/src/test/java/org/apache/samza/coordinator/TestJobModelCalculator.java