private LocalJobPlanner createLocalJobPlanner(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc) throws  Exception {
    CoordinationUtils coordinationUtils = mock(CoordinationUtils.class);
    DistributedLock distributedLock = mock(DistributedLock.class);
    when(distributedLock.lock(anyObject())).thenReturn(true);
    when(coordinationUtils.getLock(anyString())).thenReturn(distributedLock);

    ZkMetadataStore zkMetadataStore = mock(ZkMetadataStore.class);
    when(zkMetadataStore.get(any())).thenReturn(null);
    PowerMockito.whenNew(ZkMetadataStore.class).withAnyArguments().thenReturn(zkMetadataStore);

    return spy(new LocalJobPlanner(appDesc, coordinationUtils, "FAKE_UID", "FAKE_RUNID"));
  }
A test case to verify plan results in different hash in case of different intermediate stream.
Note: the overall JOB PLAN remains the same outside the scope of intermediate streams the sake of these test cases.
samza-core/src/test/java/org/apache/samza/execution/TestLocalJobPlanner.java