BaseKeyValueStorageEngineFactory implements StorageEngineFactory<K, V> { public StorageEngine getStorageEngine(String storeName, File storeDir, Serde<K> keySerde, Serde<V> msgSerde, MessageCollector changelogCollector, MetricsRegistry registry, SystemStreamPartition changelogSSP, JobContext jobContext, ContainerContext containerContext, StoreMode storeMode) { Config storageConfigSubset = jobContext.getConfig().subset("stores." + storeName + ".", true); StorageConfig storageConfig = new StorageConfig(jobContext.getConfig()); Optional<String> storeFactory = storageConfig.getStorageFactoryClassName(storeName); StoreProperties.StorePropertiesBuilder storePropertiesBuilder = new StoreProperties.StorePropertiesBuilder(); if (!storeFactory.isPresent() || StringUtils.isBlank(storeFactory.get())) { throw new SamzaException( String.format("Store factory not defined for store %s. Cannot proceed with KV store creation!", storeName)); } if (!storeFactory.get().equals(INMEMORY_KV_STORAGE_ENGINE_FACTORY)) { storePropertiesBuilder.setPersistedToDisk(true); } int batchSize = storageConfigSubset.getInt(WRITE_BATCH_SIZE, DEFAULT_WRITE_BATCH_SIZE); int cacheSize = storageConfigSubset.getInt(OBJECT_CACHE_SIZE, Math.max(batchSize, DEFAULT_OBJECT_CACHE_SIZE)); if (cacheSize > 0 && cacheSize < batchSize) { throw new SamzaException( String.format("cache.size for store %s cannot be less than batch.size as batched values reside in cache.", storeName)); } if (keySerde == null) { throw new SamzaException( String.format("Must define a key serde when using key value storage for store %s.", storeName)); } if (msgSerde == null) { throw new SamzaException( String.format("Must define a message serde when using key value storage for store %s.", storeName)); } KeyValueStore<byte[], byte[]> rawStore = getKVStore(storeName, storeDir, registry, changelogSSP, jobContext, containerContext, storeMode); KeyValueStore<byte[], byte[]> maybeLoggedStore = buildMaybeLoggedStore(changelogSSP, storeName, registry, storePropertiesBuilder, rawStore, changelogCollector); KeyValueStore<K, V> toBeAccessLoggedStore = buildStoreWithLargeMessageHandling(storeName, registry, maybeLoggedStore, storageConfig, cacheSize, batchSize, keySerde, msgSerde); KeyValueStore<K, V> maybeAccessLoggedStore = buildMaybeAccessLoggedStore(storeName, toBeAccessLoggedStore, changelogCollector, changelogSSP, storageConfig, keySerde); KeyValueStore<K, V> nullSafeStore = new NullSafeKeyValueStore<>(maybeAccessLoggedStore); KeyValueStorageEngineMetrics keyValueStorageEngineMetrics = new KeyValueStorageEngineMetrics(storeName, registry); HighResolutionClock clock = buildClock(jobContext.getConfig()); return new KeyValueStorageEngine<>(storeName, storeDir, storePropertiesBuilder.build(), nullSafeStore, rawStore, changelogSSP, changelogCollector, keyValueStorageEngineMetrics, batchSize, ScalaJavaUtil.toScalaFunction(clock::nanoTime)); }  StorageEngine getStorageEngine(String storeName,
      File storeDir,
      Serde<K> keySerde,
      Serde<V> msgSerde,
      MessageCollector changelogCollector,
      MetricsRegistry registry,
      SystemStreamPartition changelogSSP,
      JobContext jobContext,
      ContainerContext containerContext,
      StoreMode storeMode);  }
@Test(expected = SamzaException.class) public void testMissingKeySerde() { Config config = new MapConfig(BASE_CONFIG); when(this.jobContext.getConfig()).thenReturn(config); new MockKeyValueStorageEngineFactory(this.rawKeyValueStore).getStorageEngine(STORE_NAME, this.storeDir, null, this.msgSerde, this.changelogCollector, this.metricsRegistry, null, this.jobContext, this.containerContext, STORE_MODE); }
@Test(expected = SamzaException.class) public void testMissingValueSerde() { Config config = new MapConfig(BASE_CONFIG); when(this.jobContext.getConfig()).thenReturn(config); new MockKeyValueStorageEngineFactory(this.rawKeyValueStore).getStorageEngine(STORE_NAME, this.storeDir, this.keySerde, null, this.changelogCollector, this.metricsRegistry, null, this.jobContext, this.containerContext, STORE_MODE); }