private void getChangeLogSystemStreamsAndStorageFactories() {
    StorageConfig config = new StorageConfig(jobConfig);
    List<String> storeNames = config.getStoreNames();

    LOG.info("Got store names: " + storeNames.toString());

    for (String storeName : storeNames) {
      Optional<String> streamName = config.getChangelogStream(storeName);

      LOG.info("stream name for " + storeName + " is " + streamName.orElse(null));

      streamName.ifPresent(name -> changeLogSystemStreams.put(storeName, StreamUtil.getSystemStreamFromNames(name)));

      Optional<String> factoryClass = config.getStorageFactoryClassName(storeName);
      if (factoryClass.isPresent()) {
        @SuppressWarnings("unchecked")
        StorageEngineFactory<Object, Object> factory =
            (StorageEngineFactory<Object, Object>) ReflectionUtil.getObj(factoryClass.get(), StorageEngineFactory.class);

        storageEngineFactories.put(storeName, factory);
      } else {
        throw new SamzaException("Missing storage factory for " + storeName + ".");
      }
    }
  }
Get the changelog streams and the storage factories from the config file
and put them into the maps
samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java