@Test
  public void testCreateCoordinatorStreamWithCoordinatorFactory() throws Exception {
    CoordinatorStreamStore coordinatorStreamStore = mock(CoordinatorStreamStore.class);
    CoordinatorStreamMetadataStoreFactory coordinatorStreamMetadataStoreFactory = mock(CoordinatorStreamMetadataStoreFactory.class);
    doReturn(coordinatorStreamStore).when(coordinatorStreamMetadataStoreFactory).getMetadataStore(anyString(), any(Config.class), any(
        MetricsRegistry.class));
    SystemAdmins systemAdmins = mock(SystemAdmins.class);
    PowerMockito.whenNew(SystemAdmins.class).withAnyArguments().thenReturn(systemAdmins);
    LocalApplicationRunner localApplicationRunner =
        spy(new LocalApplicationRunner(mockApp, config, coordinatorStreamMetadataStoreFactory));

    // create store only if successful in creating the underlying coordinator stream
    doReturn(true).when(localApplicationRunner).createUnderlyingCoordinatorStream(eq(config));
    assertEquals(coordinatorStreamStore, localApplicationRunner.createCoordinatorStreamStore(config));
    verify(localApplicationRunner).createUnderlyingCoordinatorStream(eq(config));

    // do not create store if creating the underlying coordinator stream fails
    doReturn(false).when(localApplicationRunner).createUnderlyingCoordinatorStream(eq(config));
    assertNull(localApplicationRunner.createCoordinatorStreamStore(config));
  }
Underlying coordinator stream should be created if using CoordinatorStreamMetadataStoreFactory
samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java