@Test public void shouldCreateInMemoryStoreSupplierWithLoggedConfig() throws Exception { final StateStoreSupplier supplier = Stores.create("store") .withKeys(Serdes.String()) .withValues(Serdes.String()) .inMemory() .enableLogging(Collections.singletonMap("retention.ms", "1000")) .build(); final Map<String, String> config = supplier.logConfig(); assertTrue(supplier.loggingEnabled()); assertEquals("1000", config.get("retention.ms")); }
public static StoreFactory create(final String name) { return new StoreFactory() { @Override public <K> ValueFactory<K> withKeys(final Serde<K> keySerde) { return new ValueFactory<K>() { @Override public <V> KeyValueFactory<K, V> withValues(final Serde<V> valueSerde) { return new KeyValueFactory<K, V>() { @Override public InMemoryKeyValueFactory<K, V> inMemory() { return new InMemoryKeyValueFactory<K, V>() { private int capacity = Integer.MAX_VALUE; private final Map<String, String> logConfig = new HashMap<>(); private boolean logged = true; @Override public InMemoryKeyValueFactory<K, V> maxEntries(int capacity) { if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive"); this.capacity = capacity; return this; } @Override public InMemoryKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public InMemoryKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public StateStoreSupplier build() { log.trace("Creating InMemory Store name={} capacity={} logged={}", name, capacity, logged); if (capacity < Integer.MAX_VALUE) { return new InMemoryLRUCacheStoreSupplier<>(name, capacity, keySerde, valueSerde, logged, logConfig); } return new InMemoryKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig); } }; } @Override public PersistentKeyValueFactory<K, V> persistent() { return new PersistentKeyValueFactory<K, V>() { public boolean cachingEnabled; private long windowSize; private final Map<String, String> logConfig = new HashMap<>(); private int numSegments = 0; private long retentionPeriod = 0L; private boolean retainDuplicates = false; private boolean sessionWindows; private boolean logged = true; @Override public PersistentKeyValueFactory<K, V> windowed(final long windowSize, final long retentionPeriod, final int numSegments, final boolean retainDuplicates) { this.windowSize = windowSize; this.numSegments = numSegments; this.retentionPeriod = retentionPeriod; this.retainDuplicates = retainDuplicates; this.sessionWindows = false; return this; } @Override public PersistentKeyValueFactory<K, V> sessionWindowed(final long retentionPeriod) { this.sessionWindows = true; this.retentionPeriod = retentionPeriod; return this; } @Override public PersistentKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public PersistentKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public PersistentKeyValueFactory<K, V> enableCaching() { cachingEnabled = true; return this; } @Override public StateStoreSupplier build() { log.trace("Creating RocksDb Store name={} numSegments={} logged={}", name, numSegments, logged); if (sessionWindows) { return new RocksDBSessionStoreSupplier<>(name, retentionPeriod, keySerde, valueSerde, logged, logConfig, cachingEnabled); } else if (numSegments > 0) { return new RocksDBWindowStoreSupplier<>(name, retentionPeriod, numSegments, retainDuplicates, keySerde, valueSerde, windowSize, logged, logConfig, cachingEnabled); } return new RocksDBKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig, cachingEnabled); } }; } }; } }; } }; }
Stores { public static StoreFactory create(final String name) { return new StoreFactory() { @Override public <K> ValueFactory<K> withKeys(final Serde<K> keySerde) { return new ValueFactory<K>() { @Override public <V> KeyValueFactory<K, V> withValues(final Serde<V> valueSerde) { return new KeyValueFactory<K, V>() { @Override public InMemoryKeyValueFactory<K, V> inMemory() { return new InMemoryKeyValueFactory<K, V>() { private int capacity = Integer.MAX_VALUE; private final Map<String, String> logConfig = new HashMap<>(); private boolean logged = true; @Override public InMemoryKeyValueFactory<K, V> maxEntries(int capacity) { if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive"); this.capacity = capacity; return this; } @Override public InMemoryKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public InMemoryKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public StateStoreSupplier build() { log.trace("Creating InMemory Store name={} capacity={} logged={}", name, capacity, logged); if (capacity < Integer.MAX_VALUE) { return new InMemoryLRUCacheStoreSupplier<>(name, capacity, keySerde, valueSerde, logged, logConfig); } return new InMemoryKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig); } }; } @Override public PersistentKeyValueFactory<K, V> persistent() { return new PersistentKeyValueFactory<K, V>() { public boolean cachingEnabled; private long windowSize; private final Map<String, String> logConfig = new HashMap<>(); private int numSegments = 0; private long retentionPeriod = 0L; private boolean retainDuplicates = false; private boolean sessionWindows; private boolean logged = true; @Override public PersistentKeyValueFactory<K, V> windowed(final long windowSize, final long retentionPeriod, final int numSegments, final boolean retainDuplicates) { this.windowSize = windowSize; this.numSegments = numSegments; this.retentionPeriod = retentionPeriod; this.retainDuplicates = retainDuplicates; this.sessionWindows = false; return this; } @Override public PersistentKeyValueFactory<K, V> sessionWindowed(final long retentionPeriod) { this.sessionWindows = true; this.retentionPeriod = retentionPeriod; return this; } @Override public PersistentKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public PersistentKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public PersistentKeyValueFactory<K, V> enableCaching() { cachingEnabled = true; return this; } @Override public StateStoreSupplier build() { log.trace("Creating RocksDb Store name={} numSegments={} logged={}", name, numSegments, logged); if (sessionWindows) { return new RocksDBSessionStoreSupplier<>(name, retentionPeriod, keySerde, valueSerde, logged, logConfig, cachingEnabled); } else if (numSegments > 0) { return new RocksDBWindowStoreSupplier<>(name, retentionPeriod, numSegments, retainDuplicates, keySerde, valueSerde, windowSize, logged, logConfig, cachingEnabled); } return new RocksDBKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig, cachingEnabled); } }; } }; } }; } }; } }
Stores { public static StoreFactory create(final String name) { return new StoreFactory() { @Override public <K> ValueFactory<K> withKeys(final Serde<K> keySerde) { return new ValueFactory<K>() { @Override public <V> KeyValueFactory<K, V> withValues(final Serde<V> valueSerde) { return new KeyValueFactory<K, V>() { @Override public InMemoryKeyValueFactory<K, V> inMemory() { return new InMemoryKeyValueFactory<K, V>() { private int capacity = Integer.MAX_VALUE; private final Map<String, String> logConfig = new HashMap<>(); private boolean logged = true; @Override public InMemoryKeyValueFactory<K, V> maxEntries(int capacity) { if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive"); this.capacity = capacity; return this; } @Override public InMemoryKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public InMemoryKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public StateStoreSupplier build() { log.trace("Creating InMemory Store name={} capacity={} logged={}", name, capacity, logged); if (capacity < Integer.MAX_VALUE) { return new InMemoryLRUCacheStoreSupplier<>(name, capacity, keySerde, valueSerde, logged, logConfig); } return new InMemoryKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig); } }; } @Override public PersistentKeyValueFactory<K, V> persistent() { return new PersistentKeyValueFactory<K, V>() { public boolean cachingEnabled; private long windowSize; private final Map<String, String> logConfig = new HashMap<>(); private int numSegments = 0; private long retentionPeriod = 0L; private boolean retainDuplicates = false; private boolean sessionWindows; private boolean logged = true; @Override public PersistentKeyValueFactory<K, V> windowed(final long windowSize, final long retentionPeriod, final int numSegments, final boolean retainDuplicates) { this.windowSize = windowSize; this.numSegments = numSegments; this.retentionPeriod = retentionPeriod; this.retainDuplicates = retainDuplicates; this.sessionWindows = false; return this; } @Override public PersistentKeyValueFactory<K, V> sessionWindowed(final long retentionPeriod) { this.sessionWindows = true; this.retentionPeriod = retentionPeriod; return this; } @Override public PersistentKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public PersistentKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public PersistentKeyValueFactory<K, V> enableCaching() { cachingEnabled = true; return this; } @Override public StateStoreSupplier build() { log.trace("Creating RocksDb Store name={} numSegments={} logged={}", name, numSegments, logged); if (sessionWindows) { return new RocksDBSessionStoreSupplier<>(name, retentionPeriod, keySerde, valueSerde, logged, logConfig, cachingEnabled); } else if (numSegments > 0) { return new RocksDBWindowStoreSupplier<>(name, retentionPeriod, numSegments, retainDuplicates, keySerde, valueSerde, windowSize, logged, logConfig, cachingEnabled); } return new RocksDBKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig, cachingEnabled); } }; } }; } }; } }; }  }
Stores { public static StoreFactory create(final String name) { return new StoreFactory() { @Override public <K> ValueFactory<K> withKeys(final Serde<K> keySerde) { return new ValueFactory<K>() { @Override public <V> KeyValueFactory<K, V> withValues(final Serde<V> valueSerde) { return new KeyValueFactory<K, V>() { @Override public InMemoryKeyValueFactory<K, V> inMemory() { return new InMemoryKeyValueFactory<K, V>() { private int capacity = Integer.MAX_VALUE; private final Map<String, String> logConfig = new HashMap<>(); private boolean logged = true; @Override public InMemoryKeyValueFactory<K, V> maxEntries(int capacity) { if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive"); this.capacity = capacity; return this; } @Override public InMemoryKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public InMemoryKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public StateStoreSupplier build() { log.trace("Creating InMemory Store name={} capacity={} logged={}", name, capacity, logged); if (capacity < Integer.MAX_VALUE) { return new InMemoryLRUCacheStoreSupplier<>(name, capacity, keySerde, valueSerde, logged, logConfig); } return new InMemoryKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig); } }; } @Override public PersistentKeyValueFactory<K, V> persistent() { return new PersistentKeyValueFactory<K, V>() { public boolean cachingEnabled; private long windowSize; private final Map<String, String> logConfig = new HashMap<>(); private int numSegments = 0; private long retentionPeriod = 0L; private boolean retainDuplicates = false; private boolean sessionWindows; private boolean logged = true; @Override public PersistentKeyValueFactory<K, V> windowed(final long windowSize, final long retentionPeriod, final int numSegments, final boolean retainDuplicates) { this.windowSize = windowSize; this.numSegments = numSegments; this.retentionPeriod = retentionPeriod; this.retainDuplicates = retainDuplicates; this.sessionWindows = false; return this; } @Override public PersistentKeyValueFactory<K, V> sessionWindowed(final long retentionPeriod) { this.sessionWindows = true; this.retentionPeriod = retentionPeriod; return this; } @Override public PersistentKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public PersistentKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public PersistentKeyValueFactory<K, V> enableCaching() { cachingEnabled = true; return this; } @Override public StateStoreSupplier build() { log.trace("Creating RocksDb Store name={} numSegments={} logged={}", name, numSegments, logged); if (sessionWindows) { return new RocksDBSessionStoreSupplier<>(name, retentionPeriod, keySerde, valueSerde, logged, logConfig, cachingEnabled); } else if (numSegments > 0) { return new RocksDBWindowStoreSupplier<>(name, retentionPeriod, numSegments, retainDuplicates, keySerde, valueSerde, windowSize, logged, logConfig, cachingEnabled); } return new RocksDBKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig, cachingEnabled); } }; } }; } }; } }; }  static StoreFactory create(final String name); }
Stores { public static StoreFactory create(final String name) { return new StoreFactory() { @Override public <K> ValueFactory<K> withKeys(final Serde<K> keySerde) { return new ValueFactory<K>() { @Override public <V> KeyValueFactory<K, V> withValues(final Serde<V> valueSerde) { return new KeyValueFactory<K, V>() { @Override public InMemoryKeyValueFactory<K, V> inMemory() { return new InMemoryKeyValueFactory<K, V>() { private int capacity = Integer.MAX_VALUE; private final Map<String, String> logConfig = new HashMap<>(); private boolean logged = true; @Override public InMemoryKeyValueFactory<K, V> maxEntries(int capacity) { if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive"); this.capacity = capacity; return this; } @Override public InMemoryKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public InMemoryKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public StateStoreSupplier build() { log.trace("Creating InMemory Store name={} capacity={} logged={}", name, capacity, logged); if (capacity < Integer.MAX_VALUE) { return new InMemoryLRUCacheStoreSupplier<>(name, capacity, keySerde, valueSerde, logged, logConfig); } return new InMemoryKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig); } }; } @Override public PersistentKeyValueFactory<K, V> persistent() { return new PersistentKeyValueFactory<K, V>() { public boolean cachingEnabled; private long windowSize; private final Map<String, String> logConfig = new HashMap<>(); private int numSegments = 0; private long retentionPeriod = 0L; private boolean retainDuplicates = false; private boolean sessionWindows; private boolean logged = true; @Override public PersistentKeyValueFactory<K, V> windowed(final long windowSize, final long retentionPeriod, final int numSegments, final boolean retainDuplicates) { this.windowSize = windowSize; this.numSegments = numSegments; this.retentionPeriod = retentionPeriod; this.retainDuplicates = retainDuplicates; this.sessionWindows = false; return this; } @Override public PersistentKeyValueFactory<K, V> sessionWindowed(final long retentionPeriod) { this.sessionWindows = true; this.retentionPeriod = retentionPeriod; return this; } @Override public PersistentKeyValueFactory<K, V> enableLogging(final Map<String, String> config) { logged = true; logConfig.putAll(config); return this; } @Override public PersistentKeyValueFactory<K, V> disableLogging() { logged = false; logConfig.clear(); return this; } @Override public PersistentKeyValueFactory<K, V> enableCaching() { cachingEnabled = true; return this; } @Override public StateStoreSupplier build() { log.trace("Creating RocksDb Store name={} numSegments={} logged={}", name, numSegments, logged); if (sessionWindows) { return new RocksDBSessionStoreSupplier<>(name, retentionPeriod, keySerde, valueSerde, logged, logConfig, cachingEnabled); } else if (numSegments > 0) { return new RocksDBWindowStoreSupplier<>(name, retentionPeriod, numSegments, retainDuplicates, keySerde, valueSerde, windowSize, logged, logConfig, cachingEnabled); } return new RocksDBKeyValueStoreSupplier<>(name, keySerde, valueSerde, logged, logConfig, cachingEnabled); } }; } }; } }; } }; }  static StoreFactory create(final String name);  }