diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java index 109160de50736..27a3d488918d6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java @@ -25,8 +25,8 @@ import org.apache.kafka.streams.processor.api.Record; import org.apache.kafka.streams.processor.api.RecordMetadata; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -63,7 +63,7 @@ public void enableSendingOldValues() { private class KStreamAggregateProcessor extends ContextualProcessor> { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private Sensor droppedRecordsSensor; private TimestampedTupleForwarder tupleForwarder; @@ -74,9 +74,9 @@ public void init(final ProcessorContext> context) { Thread.currentThread().getName(), context.taskId().toString(), (StreamsMetricsImpl) context.metrics()); - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -118,7 +118,7 @@ public void process(final Record record) { newAgg = aggregator.apply(record.key(), record.value(), oldAgg); - store.put(record.key(), ValueAndTimestamp.make(newAgg, newTimestamp)); + store.put(record.key(), newAgg, newTimestamp); tupleForwarder.maybeForward( record.withValue(new Change<>(newAgg, sendOldValues ? oldAgg : null)) .withTimestamp(newTimestamp)); @@ -141,11 +141,11 @@ public String[] storeNames() { } private class KStreamAggregateValueGetter implements KTableValueGetter { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; @Override public void init(final ProcessorContext context) { - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java index e936855aac99f..b801d2b60ea5a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java @@ -24,8 +24,8 @@ import org.apache.kafka.streams.processor.api.Record; import org.apache.kafka.streams.processor.api.RecordMetadata; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,7 +58,7 @@ public void enableSendingOldValues() { private class KStreamReduceProcessor extends ContextualProcessor> { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; private Sensor droppedRecordsSensor; @@ -70,9 +70,9 @@ public void init(final ProcessorContext> context) { context.taskId().toString(), (StreamsMetricsImpl) context.metrics() ); - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -112,7 +112,7 @@ public void process(final Record record) { newTimestamp = Math.max(record.timestamp(), oldAggAndTimestamp.timestamp()); } - store.put(record.key(), ValueAndTimestamp.make(newAgg, newTimestamp)); + store.put(record.key(), newAgg, newTimestamp); tupleForwarder.maybeForward( record.withValue(new Change<>(newAgg, sendOldValues ? oldAgg : null)) .withTimestamp(newTimestamp)); @@ -136,11 +136,11 @@ public String[] storeNames() { private class KStreamReduceValueGetter implements KTableValueGetter { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; @Override public void init(final ProcessorContext context) { - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java index f10af46740495..75442555cc2d4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java @@ -22,8 +22,8 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull; @@ -60,15 +60,15 @@ public Processor, KIn, Change> get() { } private class KTableAggregateProcessor implements Processor, KIn, Change> { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; @SuppressWarnings("unchecked") @Override public void init(final ProcessorContext> context) { - store = (TimestampedKeyValueStore) context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -116,7 +116,7 @@ public void process(final Record> record) { } // update the store with the new value - store.put(record.key(), ValueAndTimestamp.make(newAgg, newTimestamp)); + store.put(record.key(), newAgg, newTimestamp); tupleForwarder.maybeForward( record.withValue(new Change<>(newAgg, sendOldValues ? oldAgg : null)) .withTimestamp(newTimestamp)); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java index 99f50be1af27d..6287aa40c55d2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java @@ -20,8 +20,8 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull; @@ -88,16 +88,16 @@ private ValueAndTimestamp computeValue(final KIn key, final ValueAndTimesta private class KTableFilterProcessor implements Processor, KIn, Change> { private ProcessorContext> context; - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; @Override public void init(final ProcessorContext> context) { this.context = context; if (queryableName != null) { - store = context.getStateStore(queryableName); + store = new KeyValueStoreWrapper<>(context, queryableName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -117,7 +117,7 @@ public void process(final Record> record) { } if (queryableName != null) { - store.put(key, ValueAndTimestamp.make(newValue, record.timestamp())); + store.put(key, newValue, record.timestamp()); tupleForwarder.maybeForward(record.withValue(new Change<>(newValue, oldValue))); } else { context.forward(record.withValue(new Change<>(newValue, oldValue))); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java index e34ac2f584130..c2786257a74de 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java @@ -175,7 +175,7 @@ private KTable doFilter(final Predicate predicate, final Serde keySerde; final Serde valueSerde; final String queryableStoreName; - final StoreBuilder> storeBuilder; + final StoreBuilder storeBuilder; if (materializedInternal != null) { // we actually do not need to generate store names at all since if it is not specified, we will not @@ -290,7 +290,7 @@ private KTable doMapValues(final ValueMapperWithKey keySerde; final Serde valueSerde; final String queryableStoreName; - final StoreBuilder> storeBuilder; + final StoreBuilder storeBuilder; if (materializedInternal != null) { // we actually do not need to generate store names at all since if it is not specified, we will not @@ -445,7 +445,7 @@ private KTable doTransformValues(final ValueTransformerWithKeySuppli final Serde keySerde; final Serde valueSerde; final String queryableStoreName; - final StoreBuilder> storeBuilder; + final StoreBuilder storeBuilder; if (materializedInternal != null) { // don't inherit parent value serde, since this operation may change the value type, more specifically: @@ -750,7 +750,7 @@ private KTable doJoin(final KTable other, final Serde keySerde; final Serde valueSerde; final String queryableStoreName; - final StoreBuilder> storeBuilder; + final StoreBuilder storeBuilder; if (materializedInternal != null) { if (materializedInternal.keySerde() == null) { @@ -1270,7 +1270,7 @@ private KTable doJoinOnForeignKey(final KTable forei materializedInternal.queryableStoreName() ); - final StoreBuilder> resultStore = + final StoreBuilder resultStore = new TimestampedKeyValueStoreMaterializer<>(materializedInternal).materialize(); final TableProcessorNode resultNode = new TableProcessorNode<>( diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinMerger.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinMerger.java index 72dbe40b7f175..2ed1dc42a979d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinMerger.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinMerger.java @@ -20,12 +20,11 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; -import org.apache.kafka.streams.state.ValueAndTimestamp; import java.util.Collections; import java.util.HashSet; import java.util.Set; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; public class KTableKTableJoinMerger implements KTableProcessorSupplier { @@ -98,7 +97,7 @@ public static KTableKTableJoinMerger of(final KTableProcessorSuppli } private class KTableKTableJoinMergeProcessor extends ContextualProcessor, K, Change> { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; @SuppressWarnings("unchecked") @@ -106,9 +105,9 @@ private class KTableKTableJoinMergeProcessor extends ContextualProcessor> context) { super.init(context); if (queryableName != null) { - store = (TimestampedKeyValueStore) context.getStateStore(queryableName); + store = new KeyValueStoreWrapper<>(context, queryableName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -118,7 +117,7 @@ public void init(final ProcessorContext> context) { @Override public void process(final Record> record) { if (queryableName != null) { - store.put(record.key(), ValueAndTimestamp.make(record.value().newValue, record.timestamp())); + store.put(record.key(), record.value().newValue, record.timestamp()); tupleForwarder.maybeForward(record); } else { if (sendOldValues) { diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java index acea6c0cf2c38..f0897dcdb050c 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java @@ -20,8 +20,8 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull; @@ -106,16 +106,16 @@ private ValueAndTimestamp computeValueAndTimestamp(final KIn key, final Va private class KTableMapValuesProcessor implements Processor, KIn, Change> { private ProcessorContext> context; - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; @Override public void init(final ProcessorContext> context) { this.context = context; if (queryableName != null) { - store = context.getStateStore(queryableName); + store = new KeyValueStoreWrapper<>(context, queryableName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -128,7 +128,7 @@ public void process(final Record> record) { final VOut oldValue = computeOldValue(record.key(), record.value()); if (queryableName != null) { - store.put(record.key(), ValueAndTimestamp.make(newValue, record.timestamp())); + store.put(record.key(), newValue, record.timestamp()); tupleForwarder.maybeForward(record.withValue(new Change<>(newValue, oldValue))); } else { context.forward(record.withValue(new Change<>(newValue, oldValue))); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java index e7570366f2454..ba7e65081d583 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java @@ -17,8 +17,8 @@ package org.apache.kafka.streams.kstream.internals; import org.apache.kafka.streams.processor.api.ProcessorContext; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; public class KTableMaterializedValueGetterSupplier implements KTableValueGetterSupplier { private final String storeName; @@ -37,11 +37,11 @@ public String[] storeNames() { } private class KTableMaterializedValueGetter implements KTableValueGetter { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; @Override public void init(final ProcessorContext context) { - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTablePassThrough.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTablePassThrough.java index a28b973d75ff6..91fe0e4a27795 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTablePassThrough.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTablePassThrough.java @@ -19,10 +19,10 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; import java.util.Collection; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; public class KTablePassThrough implements KTableProcessorSupplier { private final Collection parents; @@ -79,11 +79,11 @@ public void process(final Record> record) { } private class KTablePassThroughValueGetter implements KTableValueGetter { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; @Override public void init(final ProcessorContext context) { - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java index 08d04494957e3..ef40638dd2826 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java @@ -21,8 +21,8 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull; @@ -54,15 +54,15 @@ public Processor, K, Change> get() { private class KTableReduceProcessor implements Processor, K, Change> { - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; @SuppressWarnings("unchecked") @Override public void init(final ProcessorContext> context) { - store = (TimestampedKeyValueStore) context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -106,7 +106,7 @@ public void process(final Record> record) { } // update the store with the new value - store.put(record.key(), ValueAndTimestamp.make(newAgg, newTimestamp)); + store.put(record.key(), newAgg, newTimestamp); tupleForwarder.maybeForward( record.withValue(new Change<>(newAgg, sendOldValues ? oldAgg : null)) .withTimestamp(newTimestamp)); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java index 6de8ede316b86..7856c91d839e5 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java @@ -26,8 +26,8 @@ import org.apache.kafka.streams.processor.api.Record; import org.apache.kafka.streams.processor.api.RecordMetadata; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -76,7 +76,7 @@ public boolean materialized() { private class KTableSourceProcessor implements Processor> { private ProcessorContext> context; - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; private Sensor droppedRecordsSensor; @@ -88,9 +88,9 @@ public void init(final ProcessorContext> context) { droppedRecordsSensor = droppedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics); if (queryableName != null) { - store = context.getStateStore(queryableName); + store = new KeyValueStoreWrapper<>(context, queryableName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -146,7 +146,7 @@ public void process(final Record record) { } else { oldValue = null; } - store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp())); + store.put(record.key(), record.value(), record.timestamp()); tupleForwarder.maybeForward(record.withValue(new Change<>(record.value(), oldValue))); } else { context.forward(record.withValue(new Change<>(record.value(), null))); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java index 94701318fa2ab..1c3409433542a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java @@ -18,8 +18,8 @@ package org.apache.kafka.streams.kstream.internals; import org.apache.kafka.streams.processor.api.ProcessorContext; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; public class KTableSourceValueGetterSupplier implements KTableValueGetterSupplier { private final String storeName; @@ -38,10 +38,10 @@ public String[] storeNames() { } private class KTableSourceValueGetter implements KTableValueGetter { - private TimestampedKeyValueStore store = null; + private KeyValueStoreWrapper store; public void init(final ProcessorContext context) { - store = context.getStateStore(storeName); + store = new KeyValueStoreWrapper<>(context, storeName); } public ValueAndTimestamp get(final K key) { diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java index 6aac1d4183bea..5975571f0faa8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java @@ -26,10 +26,10 @@ import org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext; import org.apache.kafka.streams.processor.internals.InternalProcessorContext; import org.apache.kafka.streams.processor.internals.ProcessorRecordContext; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.ValueAndTimestamp; import java.util.Objects; +import org.apache.kafka.streams.state.internals.KeyValueStoreWrapper; import static org.apache.kafka.streams.processor.internals.RecordQueue.UNKNOWN; import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull; @@ -90,7 +90,7 @@ public boolean enableSendingOldValues(final boolean forceMaterialization) { private class KTableTransformValuesProcessor extends ContextualProcessor, K, Change> { private final ValueTransformerWithKey valueTransformer; - private TimestampedKeyValueStore store; + private KeyValueStoreWrapper store; private TimestampedTupleForwarder tupleForwarder; private KTableTransformValuesProcessor(final ValueTransformerWithKey valueTransformer) { @@ -103,9 +103,9 @@ public void init(final ProcessorContext> context) { final InternalProcessorContext> internalProcessorContext = (InternalProcessorContext>) context; valueTransformer.init(new ForwardingDisabledProcessorContext(internalProcessorContext)); if (queryableName != null) { - store = context.getStateStore(queryableName); + store = new KeyValueStoreWrapper<>(context, queryableName); tupleForwarder = new TimestampedTupleForwarder<>( - store, + store.getStore(), context, new TimestampedCacheFlushListener<>(context), sendOldValues); @@ -121,7 +121,7 @@ public void process(final Record> record) { context().forward(record.withValue(new Change<>(newValue, oldValue))); } else { final VOut oldValue = sendOldValues ? getValueOrNull(store.get(record.key())) : null; - store.put(record.key(), ValueAndTimestamp.make(newValue, record.timestamp())); + store.put(record.key(), newValue, record.timestamp()); tupleForwarder.maybeForward(record.withValue(new Change<>(newValue, oldValue))); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java index 2772a66810832..dec7b437b2c32 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java @@ -24,7 +24,6 @@ import org.apache.kafka.streams.processor.api.ProcessorSupplier; import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import java.util.Arrays; @@ -37,7 +36,7 @@ public class KTableKTableJoinNode extends BaseJoinProcessorNode valueSerde; private final String[] joinThisStoreNames; private final String[] joinOtherStoreNames; - private final StoreBuilder> storeBuilder; + private final StoreBuilder storeBuilder; KTableKTableJoinNode(final String nodeName, final ProcessorParameters, ?, ?> joinThisProcessorParameters, @@ -49,7 +48,7 @@ public class KTableKTableJoinNode extends BaseJoinProcessorNode valueSerde, final String[] joinThisStoreNames, final String[] joinOtherStoreNames, - final StoreBuilder> storeBuilder) { + final StoreBuilder storeBuilder) { super(nodeName, null, @@ -148,7 +147,7 @@ public static final class KTableKTableJoinNodeBuilder { private String[] joinThisStoreNames; private String[] joinOtherStoreNames; private String queryableStoreName; - private StoreBuilder> storeBuilder; + private StoreBuilder storeBuilder; private KTableKTableJoinNodeBuilder() { } @@ -203,7 +202,7 @@ public KTableKTableJoinNodeBuilder withQueryableStoreName(final S return this; } - public KTableKTableJoinNodeBuilder withStoreBuilder(final StoreBuilder> storeBuilder) { + public KTableKTableJoinNodeBuilder withStoreBuilder(final StoreBuilder storeBuilder) { this.storeBuilder = storeBuilder; return this; } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamToTableNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamToTableNode.java index 8c071dad229b1..6d79deb81058d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamToTableNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamToTableNode.java @@ -24,7 +24,6 @@ import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; /** * Represents a KTable convert From KStream @@ -53,7 +52,7 @@ public String toString() { @SuppressWarnings("unchecked") @Override public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { - final StoreBuilder> storeBuilder = + final StoreBuilder storeBuilder = new TimestampedKeyValueStoreMaterializer<>((MaterializedInternal>) materializedInternal).materialize(); final String processorName = processorParameters.processorName(); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableProcessorNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableProcessorNode.java index c522b5a9f3fea..5b25fb682f1c3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableProcessorNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableProcessorNode.java @@ -20,7 +20,6 @@ import org.apache.kafka.streams.kstream.internals.KTableSource; import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import java.util.Arrays; import java.util.Objects; @@ -28,19 +27,18 @@ public class TableProcessorNode extends GraphNode { private final ProcessorParameters processorParameters; - private final StoreBuilder> storeBuilder; + private final StoreBuilder storeBuilder; private final String[] storeNames; public TableProcessorNode(final String nodeName, final ProcessorParameters processorParameters, - final StoreBuilder> storeBuilder) { + final StoreBuilder storeBuilder) { this(nodeName, processorParameters, storeBuilder, null); } public TableProcessorNode(final String nodeName, final ProcessorParameters processorParameters, - // TODO KIP-300: we are enforcing this as a keyvalue store, but it should go beyond any type of stores - final StoreBuilder> storeBuilder, + final StoreBuilder storeBuilder, final String[] storeNames) { super(nodeName); this.processorParameters = processorParameters; diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java index 52413b9b61b17..e218263e9b42f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java @@ -27,7 +27,6 @@ import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; import java.util.Collections; @@ -94,9 +93,7 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { throw new IllegalStateException("A table source node must have a single topic as input"); } - // TODO: we assume source KTables can only be timestamped-key-value stores for now. - // should be expanded for other types of stores as well. - final StoreBuilder> storeBuilder = + final StoreBuilder storeBuilder = new TimestampedKeyValueStoreMaterializer<>((MaterializedInternal>) materializedInternal).materialize(); if (isGlobalKTable) { diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java new file mode 100644 index 0000000000000..9c2920570aaba --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapper.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.state.internals; + +import org.apache.kafka.streams.errors.InvalidStateStoreException; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.query.Position; +import org.apache.kafka.streams.query.PositionBound; +import org.apache.kafka.streams.query.Query; +import org.apache.kafka.streams.query.QueryConfig; +import org.apache.kafka.streams.query.QueryResult; +import org.apache.kafka.streams.state.TimestampedKeyValueStore; +import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * A wrapper class for non-windowed key-value stores used within the DSL. All such stores are + * instances of either {@link TimestampedKeyValueStore} or {@link VersionedKeyValueStore}. + * + * @param The key type + * @param The value type + */ +public class KeyValueStoreWrapper implements StateStore { + + private TimestampedKeyValueStore timestampedStore = null; + private VersionedKeyValueStore versionedStore = null; + + // same as either timestampedStore or versionedStore above. kept merely as a convenience + // to simplify implementation for methods which do not depend on store type. + private StateStore store = null; + + public KeyValueStoreWrapper(final ProcessorContext context, final String storeName) { + try { + // first try timestamped store + timestampedStore = context.getStateStore(storeName); + store = timestampedStore; + return; + } catch (final ClassCastException e) { + // ignore since could be versioned store instead + } + + try { + // next try versioned store + versionedStore = context.getStateStore(storeName); + store = versionedStore; + } catch (final ClassCastException e) { + store = context.getStateStore(storeName); + final String storeType = store == null ? "null" : store.getClass().getName(); + throw new InvalidStateStoreException("KTable source state store must implement either " + + "TimestampedKeyValueStore or VersionedKeyValueStore. Got: " + storeType); + } + } + + public ValueAndTimestamp get(final K key) { + if (timestampedStore != null) { + return timestampedStore.get(key); + } + if (versionedStore != null) { + final VersionedRecord versionedRecord = versionedStore.get(key); + return versionedRecord == null + ? null + : ValueAndTimestamp.make(versionedRecord.value(), versionedRecord.timestamp()); + } + throw new IllegalStateException("KeyValueStoreWrapper must be initialized with either timestamped or versioned store"); + } + + public void put(final K key, final V value, final long timestamp) { + if (timestampedStore != null) { + timestampedStore.put(key, ValueAndTimestamp.make(value, timestamp)); + return; + } + if (versionedStore != null) { + versionedStore.put(key, value, timestamp); + return; + } + throw new IllegalStateException("KeyValueStoreWrapper must be initialized with either timestamped or versioned store"); + } + + public StateStore getStore() { + return store; + } + + @Override + public String name() { + return store.name(); + } + + @Deprecated + @Override + public void init(final org.apache.kafka.streams.processor.ProcessorContext context, final StateStore root) { + store.init(context, root); + } + + @Override + public void init(final StateStoreContext context, final StateStore root) { + store.init(context, root); + } + + @Override + public void flush() { + store.flush(); + } + + @Override + public void close() { + store.close(); + } + + @Override + public boolean persistent() { + return store.persistent(); + } + + @Override + public boolean isOpen() { + return store.isOpen(); + } + + @Override + public QueryResult query(final Query query, final PositionBound positionBound, final QueryConfig config) { + return store.query(query, positionBound, config); + } + + @Override + public Position getPosition() { + return store.getPosition(); + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapperTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapperTest.java new file mode 100644 index 0000000000000..869d4858f7572 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreWrapperTest.java @@ -0,0 +1,356 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.state.internals; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.nullValue; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.kafka.streams.errors.InvalidStateStoreException; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.query.Position; +import org.apache.kafka.streams.query.PositionBound; +import org.apache.kafka.streams.query.Query; +import org.apache.kafka.streams.query.QueryConfig; +import org.apache.kafka.streams.query.QueryResult; +import org.apache.kafka.streams.state.KeyValueStore; +import org.apache.kafka.streams.state.TimestampedKeyValueStore; +import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.StrictStubs.class) +public class KeyValueStoreWrapperTest { + + private static final String STORE_NAME = "kvStore"; + private static final String KEY = "k"; + private static final ValueAndTimestamp VALUE_AND_TIMESTAMP + = ValueAndTimestamp.make("v", 8L); + + @Mock + private TimestampedKeyValueStore timestampedStore; + @Mock + private VersionedKeyValueStore versionedStore; + @Mock + private ProcessorContext context; + @Mock + private Query query; + @Mock + private PositionBound positionBound; + @Mock + private QueryConfig queryConfig; + @Mock + private QueryResult result; + @Mock + private Position position; + + private KeyValueStoreWrapper wrapper; + + @Test + public void shouldThrowOnNonTimestampedOrVersionedStore() { + when(context.getStateStore(STORE_NAME)).thenReturn(mock(KeyValueStore.class)); + + assertThrows(InvalidStateStoreException.class, () -> new KeyValueStoreWrapper<>(context, STORE_NAME)); + } + + @Test + public void shouldGetFromTimestampedStore() { + givenWrapperWithTimestampedStore(); + when(timestampedStore.get(KEY)).thenReturn(VALUE_AND_TIMESTAMP); + + assertThat(wrapper.get(KEY), equalTo(VALUE_AND_TIMESTAMP)); + } + + @Test + public void shouldGetFromVersionedStore() { + givenWrapperWithVersionedStore(); + when(versionedStore.get(KEY)).thenReturn( + new VersionedRecord<>( + VALUE_AND_TIMESTAMP.value(), + VALUE_AND_TIMESTAMP.timestamp()) + ); + + assertThat(wrapper.get(KEY), equalTo(VALUE_AND_TIMESTAMP)); + } + + @Test + public void shouldGetNullFromTimestampedStore() { + givenWrapperWithTimestampedStore(); + when(timestampedStore.get(KEY)).thenReturn(null); + + assertThat(wrapper.get(KEY), nullValue()); + } + + @Test + public void shouldGetNullFromVersionedStore() { + givenWrapperWithVersionedStore(); + when(versionedStore.get(KEY)).thenReturn(null); + + assertThat(wrapper.get(KEY), nullValue()); + } + + @Test + public void shouldPutToTimestampedStore() { + givenWrapperWithTimestampedStore(); + + wrapper.put(KEY, VALUE_AND_TIMESTAMP.value(), VALUE_AND_TIMESTAMP.timestamp()); + + verify(timestampedStore).put(KEY, VALUE_AND_TIMESTAMP); + } + + @Test + public void shouldPutToVersionedStore() { + givenWrapperWithVersionedStore(); + + wrapper.put(KEY, VALUE_AND_TIMESTAMP.value(), VALUE_AND_TIMESTAMP.timestamp()); + + verify(versionedStore).put(KEY, VALUE_AND_TIMESTAMP.value(), VALUE_AND_TIMESTAMP.timestamp()); + } + + @Test + public void shouldPutNullToTimestampedStore() { + givenWrapperWithTimestampedStore(); + + wrapper.put(KEY, null, VALUE_AND_TIMESTAMP.timestamp()); + + verify(timestampedStore).put(KEY, null); + } + + @Test + public void shouldPutNullToVersionedStore() { + givenWrapperWithVersionedStore(); + + wrapper.put(KEY, null, VALUE_AND_TIMESTAMP.timestamp()); + + verify(versionedStore).put(KEY, null, VALUE_AND_TIMESTAMP.timestamp()); + } + + @Test + public void shouldGetTimestampedStore() { + givenWrapperWithTimestampedStore(); + + assertThat(wrapper.getStore(), equalTo(timestampedStore)); + } + + @Test + public void shouldGetVersionedStore() { + givenWrapperWithVersionedStore(); + + assertThat(wrapper.getStore(), equalTo(versionedStore)); + } + + @Test + public void shouldGetNameForTimestampedStore() { + givenWrapperWithTimestampedStore(); + when(timestampedStore.name()).thenReturn(STORE_NAME); + + assertThat(wrapper.name(), equalTo(STORE_NAME)); + } + + @Test + public void shouldGetNameForVersionedStore() { + givenWrapperWithVersionedStore(); + when(versionedStore.name()).thenReturn(STORE_NAME); + + assertThat(wrapper.name(), equalTo(STORE_NAME)); + } + + @Deprecated + @Test + public void shouldDeprecatedInitTimestampedStore() { + givenWrapperWithTimestampedStore(); + final org.apache.kafka.streams.processor.ProcessorContext mockContext + = mock(org.apache.kafka.streams.processor.ProcessorContext.class); + + wrapper.init(mockContext, wrapper); + + verify(timestampedStore).init(mockContext, wrapper); + } + + @Deprecated + @Test + public void shouldDeprecatedInitVersionedStore() { + givenWrapperWithVersionedStore(); + final org.apache.kafka.streams.processor.ProcessorContext mockContext + = mock(org.apache.kafka.streams.processor.ProcessorContext.class); + + wrapper.init(mockContext, wrapper); + + verify(versionedStore).init(mockContext, wrapper); + } + + @Test + public void shouldInitTimestampedStore() { + givenWrapperWithTimestampedStore(); + final StateStoreContext mockContext = mock(StateStoreContext.class); + + wrapper.init(mockContext, wrapper); + + verify(timestampedStore).init(mockContext, wrapper); + } + + @Test + public void shouldInitVersionedStore() { + givenWrapperWithVersionedStore(); + final StateStoreContext mockContext = mock(StateStoreContext.class); + + wrapper.init(mockContext, wrapper); + + verify(versionedStore).init(mockContext, wrapper); + } + + @Test + public void shouldFlushTimestampedStore() { + givenWrapperWithTimestampedStore(); + + wrapper.flush(); + + verify(timestampedStore).flush(); + } + + @Test + public void shouldFlushVersionedStore() { + givenWrapperWithVersionedStore(); + + wrapper.flush(); + + verify(versionedStore).flush(); + } + + @Test + public void shouldCloseTimestampedStore() { + givenWrapperWithTimestampedStore(); + + wrapper.close(); + + verify(timestampedStore).close(); + } + + @Test + public void shouldCloseVersionedStore() { + givenWrapperWithVersionedStore(); + + wrapper.close(); + + verify(versionedStore).close(); + } + + @Test + public void shouldReturnPersistentForTimestampedStore() { + givenWrapperWithTimestampedStore(); + + // test "persistent = true" + when(timestampedStore.persistent()).thenReturn(true); + assertThat(wrapper.persistent(), equalTo(true)); + + // test "persistent = false" + when(timestampedStore.persistent()).thenReturn(false); + assertThat(wrapper.persistent(), equalTo(false)); + } + + @Test + public void shouldReturnPersistentForVersionedStore() { + givenWrapperWithVersionedStore(); + + // test "persistent = true" + when(versionedStore.persistent()).thenReturn(true); + assertThat(wrapper.persistent(), equalTo(true)); + + // test "persistent = false" + when(versionedStore.persistent()).thenReturn(false); + assertThat(wrapper.persistent(), equalTo(false)); + } + + @Test + public void shouldReturnIsOpenForTimestampedStore() { + givenWrapperWithTimestampedStore(); + + // test "isOpen = true" + when(timestampedStore.isOpen()).thenReturn(true); + assertThat(wrapper.isOpen(), equalTo(true)); + + // test "isOpen = false" + when(timestampedStore.isOpen()).thenReturn(false); + assertThat(wrapper.isOpen(), equalTo(false)); + } + + @Test + public void shouldReturnIsOpenForVersionedStore() { + givenWrapperWithVersionedStore(); + + // test "isOpen = true" + when(versionedStore.isOpen()).thenReturn(true); + assertThat(wrapper.isOpen(), equalTo(true)); + + // test "isOpen = false" + when(versionedStore.isOpen()).thenReturn(false); + assertThat(wrapper.isOpen(), equalTo(false)); + } + + @SuppressWarnings("unchecked") + @Test + public void shouldQueryTimestampedStore() { + givenWrapperWithTimestampedStore(); + when(timestampedStore.query(query, positionBound, queryConfig)).thenReturn(result); + + assertThat(wrapper.query(query, positionBound, queryConfig), equalTo(result)); + } + + @SuppressWarnings("unchecked") + @Test + public void shouldQueryVersionedStore() { + givenWrapperWithVersionedStore(); + when(versionedStore.query(query, positionBound, queryConfig)).thenReturn(result); + + assertThat(wrapper.query(query, positionBound, queryConfig), equalTo(result)); + } + + @Test + public void shouldGetPositionForTimestampedStore() { + givenWrapperWithTimestampedStore(); + when(timestampedStore.getPosition()).thenReturn(position); + + assertThat(wrapper.getPosition(), equalTo(position)); + } + + @Test + public void shouldGetPositionForVersionedStore() { + givenWrapperWithVersionedStore(); + when(versionedStore.getPosition()).thenReturn(position); + + assertThat(wrapper.getPosition(), equalTo(position)); + } + + private void givenWrapperWithTimestampedStore() { + when(context.getStateStore(STORE_NAME)).thenReturn(timestampedStore); + wrapper = new KeyValueStoreWrapper<>(context, STORE_NAME); + } + + private void givenWrapperWithVersionedStore() { + when(context.getStateStore(STORE_NAME)).thenReturn(versionedStore); + wrapper = new KeyValueStoreWrapper<>(context, STORE_NAME); + } +} \ No newline at end of file