@SuppressWarnings("unchecked") @Test public void testAddRemove() throws Exception { context.setTime(1); changeLogger.logChange(0, "zero"); changeLogger.logChange(1, "one"); changeLogger.logChange(2, "two"); assertEquals("zero", logged.get(0)); assertEquals("one", logged.get(1)); assertEquals("two", logged.get(2)); changeLogger.logChange(0, null); assertNull(logged.get(0)); }
void logChange(final K key, final V value) { if (collector != null) { final Serializer<K> keySerializer = serialization.keySerializer(); final Serializer<V> valueSerializer = serialization.valueSerializer(); collector.send(this.topic, key, value, this.partition, context.timestamp(), keySerializer, valueSerializer); } }
StoreChangeLogger { void logChange(final K key, final V value) { if (collector != null) { final Serializer<K> keySerializer = serialization.keySerializer(); final Serializer<V> valueSerializer = serialization.valueSerializer(); collector.send(this.topic, key, value, this.partition, context.timestamp(), keySerializer, valueSerializer); } } }
StoreChangeLogger { void logChange(final K key, final V value) { if (collector != null) { final Serializer<K> keySerializer = serialization.keySerializer(); final Serializer<V> valueSerializer = serialization.valueSerializer(); collector.send(this.topic, key, value, this.partition, context.timestamp(), keySerializer, valueSerializer); } } StoreChangeLogger(String storeName, ProcessorContext context, StateSerdes<K, V> serialization); private  StoreChangeLogger(String storeName, ProcessorContext context, int partition, StateSerdes<K, V> serialization); }
StoreChangeLogger { void logChange(final K key, final V value) { if (collector != null) { final Serializer<K> keySerializer = serialization.keySerializer(); final Serializer<V> valueSerializer = serialization.valueSerializer(); collector.send(this.topic, key, value, this.partition, context.timestamp(), keySerializer, valueSerializer); } } StoreChangeLogger(String storeName, ProcessorContext context, StateSerdes<K, V> serialization); private  StoreChangeLogger(String storeName, ProcessorContext context, int partition, StateSerdes<K, V> serialization);  }
StoreChangeLogger { void logChange(final K key, final V value) { if (collector != null) { final Serializer<K> keySerializer = serialization.keySerializer(); final Serializer<V> valueSerializer = serialization.valueSerializer(); collector.send(this.topic, key, value, this.partition, context.timestamp(), keySerializer, valueSerializer); } } StoreChangeLogger(String storeName, ProcessorContext context, StateSerdes<K, V> serialization); private  StoreChangeLogger(String storeName, ProcessorContext context, int partition, StateSerdes<K, V> serialization);   }