@Test
  public void testGetBackupManagers() {
    String storeName = "store1";
    String storeName2 = "store2";
    Map<String, String> configMap = new HashMap<>();
    configMap.put(String.format(FACTORY, storeName), "store1.factory.class");
    configMap.put(String.format(FACTORY, storeName2), "store2.factory.class");

    // empty config, return no restore managers
    assertEquals(Collections.emptySet(), new StorageConfig(new MapConfig(configMap)).getBackupFactories());
    assertEquals(Collections.emptyList(), new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName));
    assertEquals(Collections.emptyList(), new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName2));

    // changelog set, should default to kafka state backend restore
    String changelogStreamOverride = "changelogStream";
    configMap.put(String.format(CHANGELOG_STREAM, storeName), changelogStreamOverride);
    configMap.put(String.format(CHANGELOG_STREAM, storeName2), changelogStreamOverride);
    configMap.put(StorageConfig.CHANGELOG_SYSTEM, "changelog-system");
    configMap.put(String.format(StorageConfig.CHANGELOG_STREAM, storeName), "changelog-stream0");
    assertEquals(ImmutableSet.of(KAFKA_STATE_BACKEND_FACTORY), new StorageConfig(new MapConfig(configMap)).getBackupFactories());
    assertEquals(DEFAULT_BACKUP_FACTORIES, new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName));
    assertEquals(DEFAULT_BACKUP_FACTORIES, new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName2));
    assertEquals(ImmutableSet.of(storeName2, storeName),
        ImmutableSet.copyOf(new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(KAFKA_STATE_BACKEND_FACTORY)));

    // job restore manager config set should override to job backend factory
    String jobBackupFactory1 = "jobBackendBackupFactory1";
    String jobBackupFactory2 = "jobBackendBackupFactory2";
    String jobBackupFactoryOverride = jobBackupFactory1 + "," + jobBackupFactory2;
    configMap.put(JOB_BACKUP_FACTORIES, jobBackupFactoryOverride);
    assertEquals(ImmutableSet.of(jobBackupFactory1, jobBackupFactory2),
        new StorageConfig(new MapConfig(configMap)).getBackupFactories());
    assertEquals(ImmutableSet.of(jobBackupFactory1, jobBackupFactory2),
        ImmutableSet.copyOf(new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName)));
    assertEquals(ImmutableSet.of(jobBackupFactory1, jobBackupFactory2),
        ImmutableSet.copyOf(new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName2)));
    assertEquals(Collections.emptyList(),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(KAFKA_STATE_BACKEND_FACTORY));
    assertEquals(ImmutableSet.of(storeName2, storeName),
        ImmutableSet.copyOf(new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(jobBackupFactory1)));
    assertEquals(ImmutableSet.of(storeName2, storeName),
        ImmutableSet.copyOf(new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(jobBackupFactory2)));

    // store specific restore managers set
    String storeBackupFactory1 = "storeBackendBackupFactory1";
    String storeBackupFactory2 = "storeBackendBackupFactory2";
    String storeBackupFactoryOverride = storeBackupFactory1 + "," + storeBackupFactory2;
    configMap.put(String.format(STORE_BACKUP_FACTORIES, storeName), storeBackupFactoryOverride);
    assertEquals(ImmutableSet.of(jobBackupFactory1, jobBackupFactory2, storeBackupFactory1, storeBackupFactory2),
        new StorageConfig(new MapConfig(configMap)).getBackupFactories());
    assertEquals(ImmutableList.of(storeBackupFactory1, storeBackupFactory2),
        new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName));
    assertEquals(ImmutableList.of(jobBackupFactory1, jobBackupFactory2),
        new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName2));
    assertEquals(Collections.emptyList(),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(KAFKA_STATE_BACKEND_FACTORY));
    assertEquals(ImmutableList.of(storeName2),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(jobBackupFactory1));
    assertEquals(ImmutableList.of(storeName2),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(jobBackupFactory2));
    assertEquals(ImmutableList.of(storeName),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(storeBackupFactory1));
    assertEquals(ImmutableList.of(storeName),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(storeBackupFactory2));

    String emptyBackupFactory = "";
    configMap.put(String.format(STORE_BACKUP_FACTORIES, storeName), emptyBackupFactory);
    assertEquals(ImmutableSet.of(jobBackupFactory1, jobBackupFactory2),
        new StorageConfig(new MapConfig(configMap)).getBackupFactories());
    assertEquals(Collections.emptyList(),
        new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName));
    assertEquals(ImmutableList.of(jobBackupFactory1, jobBackupFactory2),
        new StorageConfig(new MapConfig(configMap)).getStoreBackupFactories(storeName2));
    assertEquals(Collections.emptyList(),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(KAFKA_STATE_BACKEND_FACTORY));
    assertEquals(Collections.emptyList(),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(storeBackupFactory1));
    assertEquals(Collections.emptyList(),
        new StorageConfig(new MapConfig(configMap)).getStoresWithBackupFactory(storeBackupFactory2));
  }
Test verifies that the {@link StorageConfig#STORE_RESTORE_FACTORIES} which matches pattern for store.%s.factory
is not picked up as in store names list
samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java