private static KeyValueStore<byte[], byte[]> buildMaybeLoggedStore(SystemStreamPartition changelogSSP,
      String storeName,
      MetricsRegistry registry,
      StoreProperties.StorePropertiesBuilder storePropertiesBuilder,
      KeyValueStore<byte[], byte[]> storeToWrap,
      MessageCollector changelogCollector) {
    if (changelogSSP == null) {
      return storeToWrap;
    } else {
      LoggedStoreMetrics loggedStoreMetrics = new LoggedStoreMetrics(storeName, registry);
      storePropertiesBuilder.setLoggedStore(true);
      return new LoggedStore<>(storeToWrap, changelogSSP, changelogCollector, loggedStoreMetrics);
    }
  }
Wraps {@code storeToWrap} into a {@link LoggedStore} if {@code changelogSSP} is defined.
Otherwise, returns the original {@code storeToWrap}.
samza-kv/src/main/java/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.java