private static <T, U> KeyValueStore<T, U> buildStoreWithLargeMessageHandling(String storeName,
      MetricsRegistry registry,
      KeyValueStore<byte[], byte[]> storeToWrap,
      StorageConfig storageConfig,
      int cacheSize,
      int batchSize,
      Serde<T> keySerde,
      Serde<U> msgSerde) {
    int maxMessageSize = storageConfig.getChangelogMaxMsgSizeBytes(storeName);
    if (storageConfig.getDisallowLargeMessages(storeName)) {
      /*
       * The store wrapping ordering is done this way so that a large message cannot end up in the cache. However, it
       * also means that serialized data is in the cache, so performance will be worse since the data needs to be
       * deserialized even when cached.
       */
      KeyValueStore<byte[], byte[]> maybeCachedStore =
          buildMaybeCachedStore(storeName, registry, storeToWrap, cacheSize, batchSize);
      // this will throw a RecordTooLargeException when a large message is encountered
      LargeMessageSafeStore largeMessageSafeKeyValueStore =
          new LargeMessageSafeStore(maybeCachedStore, storeName, false, maxMessageSize);
      return buildSerializedStore(storeName, registry, largeMessageSafeKeyValueStore, keySerde, msgSerde);
    } else {
      KeyValueStore<byte[], byte[]> toBeSerializedStore;
      if (storageConfig.getDropLargeMessages(storeName)) {
        toBeSerializedStore = new LargeMessageSafeStore(storeToWrap, storeName, true, maxMessageSize);
      } else {
        toBeSerializedStore = storeToWrap;
      }
      KeyValueStore<T, U> serializedStore =
          buildSerializedStore(storeName, registry, toBeSerializedStore, keySerde, msgSerde);
      /*
       * Allows deserialized entries to be stored in the cache, but it means that a large message may end up in the
       * cache even though it was not persisted to the logged store.
       */
      return buildMaybeCachedStore(storeName, registry, serializedStore, cacheSize, batchSize);
    }
  }
Wraps {@code storeToWrap} with the proper layers to handle large messages.
If "disallow.large.messages" is enabled, then the message will be serialized and the size will be checked before
storing in the serialized message in the cache.
If "disallow.large.messages" is disabled, then the deserialized message will be stored in the cache. If
"drop.large.messages" is enabled, then large messages will not be sent to the logged store.
samza-kv/src/main/java/org/apache/samza/storage/kv/BaseKeyValueStorageEngineFactory.java