@Before
  public void setup() {

    KeyValueStore<byte[], byte[]> kvStore;
    switch (typeOfStore) {
      case "inmemory" : {
        kvStore = new InMemoryKeyValueStore(keyValueStoreMetrics);
        break;
      }
      case "rocksdb" : {
        kvStore = new RocksDbKeyValueStore(dir, new org.rocksdb.Options().setCreateIfMissing(true).setCompressionType(org.rocksdb.CompressionType.SNAPPY_COMPRESSION),
            new MapConfig(), false, storeName,
            new WriteOptions(), new FlushOptions(), keyValueStoreMetrics);
        break;
      }
      default :
        throw new IllegalArgumentException("Type of store undefined: " + typeOfStore);
    }

    MessageCollector collector = envelope -> {
      int messageLength = ((byte[]) envelope.getMessage()).length;
      if (messageLength > maxMessageSize) {
        throw new SamzaException("Logged store message size " + messageLength + " for store " + storeName
            + " was larger than the maximum allowed message size " + maxMessageSize + ".");
      }
    };
    loggedStore = new LoggedStore<>(kvStore, systemStreamPartition, collector, loggedStoreMetrics);

    switch (storeConfig) {
      case "serde" : {
        KeyValueStore<byte[], byte[]> largeMessageSafeStore =
            new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize);
        store = new SerializedKeyValueStore<>(largeMessageSafeStore, stringSerde, stringSerde,
            serializedKeyValueStoreMetrics);
        break;
      }
      case "cache-then-serde" : {
        KeyValueStore<byte[], byte[]> toBeSerializedStore = loggedStore;
        if (dropLargeMessage) {
          toBeSerializedStore = new LargeMessageSafeStore(loggedStore, storeName, dropLargeMessage, maxMessageSize);
        }
        KeyValueStore<String, String> serializedStore =
            new SerializedKeyValueStore<>(toBeSerializedStore, stringSerde, stringSerde, serializedKeyValueStoreMetrics);
        store = new CachedStore<>(serializedStore, cacheSize, batchSize, cachedStoreMetrics);
        break;
      }
      //For this case, the value of dropLargeMessage doesn't matter since we are testing the case when
      // large messages are expected and StorageConfig.DISALLOW_LARGE_MESSAGES is true.
      case "serde-then-cache" : {
        KeyValueStore<byte[], byte[]> cachedStore =
            new CachedStore<>(loggedStore, cacheSize, batchSize, cachedStoreMetrics);
        KeyValueStore<byte[], byte[]> largeMessageSafeStore =
            new LargeMessageSafeStore(cachedStore, storeName, dropLargeMessage, maxMessageSize);
        store = new SerializedKeyValueStore<>(largeMessageSafeStore, stringSerde, stringSerde,
            serializedKeyValueStoreMetrics);
        break;
      }
      default :
        throw new IllegalArgumentException("Store config undefined: " + storeConfig);
    }
    store = new NullSafeKeyValueStore<>(store);
  }
@param typeOfStore Defines type of key-value store (Eg: "rocksdb" / "inmemory")
@param storeConfig Defines what order we are invoking the stores in - serde / cache-then-serde / serde-then-cache
@param dropLargeMessageStr Defines the value of the drop.large.message config which drops large messages if true
samza-test/src/test/java/org/apache/samza/storage/kv/TestLargeMessageSafeKeyValueStores.java