From b566fd470f35b45d66743e9ba0abda5b5599531d Mon Sep 17 00:00:00 2001 From: Victoria Xia Date: Thu, 16 Feb 2023 19:50:20 -0800 Subject: [PATCH 1/2] add versioned builder --- .../internals/CogroupedKStreamImpl.java | 2 +- .../kstream/internals/KGroupedStreamImpl.java | 2 +- .../kstream/internals/KGroupedTableImpl.java | 2 +- .../streams/kstream/internals/KTableImpl.java | 10 +- ...er.java => KeyValueStoreMaterializer.java} | 37 ++- .../internals/graph/StreamToTableNode.java | 4 +- .../internals/graph/TableSourceNode.java | 4 +- .../internals/AbstractReadOnlyDecorator.java | 33 +++ .../internals/AbstractReadWriteDecorator.java | 33 +++ .../VersionedKeyValueStoreBuilder.java | 80 ++++++ .../internals/GlobalStreamThreadTest.java | 4 +- .../KeyValueStoreMaterializerTest.java | 245 ++++++++++++++++++ ...estampedKeyValueStoreMaterializerTest.java | 125 --------- .../TimestampedKeyValueStoreBuilderTest.java | 35 ++- .../VersionedKeyValueStoreBuilderTest.java | 146 +++++++++++ 15 files changed, 604 insertions(+), 158 deletions(-) rename streams/src/main/java/org/apache/kafka/streams/kstream/internals/{TimestampedKeyValueStoreMaterializer.java => KeyValueStoreMaterializer.java} (62%) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilder.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/processor/internals/KeyValueStoreMaterializerTest.java delete mode 100644 streams/src/test/java/org/apache/kafka/streams/processor/internals/TimestampedKeyValueStoreMaterializerTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilderTest.java diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java index a4a0351a88279..05b5e483a3a39 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java @@ -140,7 +140,7 @@ private KTable doAggregate(final Initializer initializer, groupPatterns, initializer, named, - new TimestampedKeyValueStoreMaterializer<>(materializedInternal).materialize(), + new KeyValueStoreMaterializer<>(materializedInternal).materialize(), materializedInternal.keySerde(), materializedInternal.valueSerde(), materializedInternal.queryableStoreName()); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java index d56caed72be82..869aeab17d795 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java @@ -238,7 +238,7 @@ private KTable doAggregate(final KStreamAggProcessorSupplier> materializedInternal) { return aggregateBuilder.build( new NamedInternal(functionName), - new TimestampedKeyValueStoreMaterializer<>(materializedInternal).materialize(), + new KeyValueStoreMaterializer<>(materializedInternal).materialize(), aggregateSupplier, materializedInternal.queryableStoreName(), materializedInternal.keySerde(), diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java index da35dac8f45cb..17cb0acd65046 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java @@ -90,7 +90,7 @@ private KTable doAggregate(final ProcessorSupplier, final StatefulProcessorNode statefulProcessorNode = new StatefulProcessorNode<>( funcName, new ProcessorParameters<>(aggregateSupplier, funcName), - new TimestampedKeyValueStoreMaterializer<>(materialized).materialize() + new KeyValueStoreMaterializer<>(materialized).materialize() ); // now the repartition node must be the parent of the StateProcessorNode 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 c2786257a74de..cd8c5abc280d1 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 @@ -190,7 +190,7 @@ private KTable doFilter(final Predicate predicate, valueSerde = materializedInternal.valueSerde() != null ? materializedInternal.valueSerde() : this.valueSerde; queryableStoreName = materializedInternal.queryableStoreName(); // only materialize if materialized is specified and it has queryable name - storeBuilder = queryableStoreName != null ? (new TimestampedKeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; + storeBuilder = queryableStoreName != null ? (new KeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; } else { keySerde = this.keySerde; valueSerde = this.valueSerde; @@ -302,7 +302,7 @@ private KTable doMapValues(final ValueMapperWithKey(materializedInternal)).materialize() : null; + storeBuilder = queryableStoreName != null ? (new KeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; } else { keySerde = this.keySerde; valueSerde = null; @@ -455,7 +455,7 @@ private KTable doTransformValues(final ValueTransformerWithKeySuppli valueSerde = materializedInternal.valueSerde(); queryableStoreName = materializedInternal.queryableStoreName(); // only materialize if materialized is specified and it has queryable name - storeBuilder = queryableStoreName != null ? (new TimestampedKeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; + storeBuilder = queryableStoreName != null ? (new KeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; } else { keySerde = this.keySerde; valueSerde = null; @@ -759,7 +759,7 @@ private KTable doJoin(final KTable other, keySerde = materializedInternal.keySerde(); valueSerde = materializedInternal.valueSerde(); queryableStoreName = materializedInternal.storeName(); - storeBuilder = new TimestampedKeyValueStoreMaterializer<>(materializedInternal).materialize(); + storeBuilder = new KeyValueStoreMaterializer<>(materializedInternal).materialize(); } else { keySerde = this.keySerde; valueSerde = null; @@ -1271,7 +1271,7 @@ private KTable doJoinOnForeignKey(final KTable forei ); final StoreBuilder resultStore = - new TimestampedKeyValueStoreMaterializer<>(materializedInternal).materialize(); + new KeyValueStoreMaterializer<>(materializedInternal).materialize(); final TableProcessorNode resultNode = new TableProcessorNode<>( resultProcessorName, diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimestampedKeyValueStoreMaterializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java similarity index 62% rename from streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimestampedKeyValueStoreMaterializer.java rename to streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java index edbde7e11c8ba..5f8440894e1bd 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimestampedKeyValueStoreMaterializer.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java @@ -17,23 +17,30 @@ package org.apache.kafka.streams.kstream.internals; import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.common.utils.Time; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; +import org.apache.kafka.streams.state.VersionedBytesStoreSupplier; +import org.apache.kafka.streams.state.internals.TimestampedKeyValueStoreBuilder; +import org.apache.kafka.streams.state.internals.VersionedKeyValueStoreBuilder; -public class TimestampedKeyValueStoreMaterializer { +/** + * Materializes a key-value store as either a {@link TimestampedKeyValueStoreBuilder} or a + * {@link VersionedKeyValueStoreBuilder} depending on whether the store is versioned or not. + */ +public class KeyValueStoreMaterializer { private final MaterializedInternal> materialized; - public TimestampedKeyValueStoreMaterializer(final MaterializedInternal> materialized) { + public KeyValueStoreMaterializer(final MaterializedInternal> materialized) { this.materialized = materialized; } /** * @return StoreBuilder */ - public StoreBuilder> materialize() { + public StoreBuilder materialize() { KeyValueBytesStoreSupplier supplier = (KeyValueBytesStoreSupplier) materialized.storeSupplier(); if (supplier == null) { switch (materialized.storeType()) { @@ -48,10 +55,19 @@ public StoreBuilder> materialize() { } } - final StoreBuilder> builder = Stores.timestampedKeyValueStoreBuilder( - supplier, - materialized.keySerde(), - materialized.valueSerde()); + final StoreBuilder builder; + if (supplier instanceof VersionedBytesStoreSupplier) { + builder = new VersionedKeyValueStoreBuilder<>( + (VersionedBytesStoreSupplier) supplier, + materialized.keySerde(), + materialized.valueSerde(), + Time.SYSTEM); + } else { + builder = Stores.timestampedKeyValueStoreBuilder( + supplier, + materialized.keySerde(), + materialized.valueSerde()); + } if (materialized.loggingEnabled()) { builder.withLoggingEnabled(materialized.logConfig()); @@ -59,9 +75,10 @@ public StoreBuilder> materialize() { builder.withLoggingDisabled(); } - if (materialized.cachingEnabled()) { + // versioned stores do not support caching + if (materialized.cachingEnabled() && !(builder instanceof VersionedKeyValueStoreBuilder)) { builder.withCachingEnabled(); } return builder; } -} +} \ No newline at end of file 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 6d79deb81058d..4825e61c4a507 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 @@ -20,7 +20,7 @@ import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.kstream.internals.KTableSource; import org.apache.kafka.streams.kstream.internals.MaterializedInternal; -import org.apache.kafka.streams.kstream.internals.TimestampedKeyValueStoreMaterializer; +import org.apache.kafka.streams.kstream.internals.KeyValueStoreMaterializer; import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; @@ -53,7 +53,7 @@ public String toString() { @Override public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { final StoreBuilder storeBuilder = - new TimestampedKeyValueStoreMaterializer<>((MaterializedInternal>) materializedInternal).materialize(); + new KeyValueStoreMaterializer<>((MaterializedInternal>) materializedInternal).materialize(); final String processorName = processorParameters.processorName(); final KTableSource tableSource = processorParameters.processorSupplier() instanceof KTableSource ? 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 e218263e9b42f..a44102119f61b 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 @@ -22,7 +22,7 @@ import org.apache.kafka.streams.kstream.internals.ConsumedInternal; import org.apache.kafka.streams.kstream.internals.KTableSource; import org.apache.kafka.streams.kstream.internals.MaterializedInternal; -import org.apache.kafka.streams.kstream.internals.TimestampedKeyValueStoreMaterializer; +import org.apache.kafka.streams.kstream.internals.KeyValueStoreMaterializer; import org.apache.kafka.streams.processor.api.ProcessorSupplier; import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; import org.apache.kafka.streams.state.KeyValueStore; @@ -94,7 +94,7 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { } final StoreBuilder storeBuilder = - new TimestampedKeyValueStoreMaterializer<>((MaterializedInternal>) materializedInternal).materialize(); + new KeyValueStoreMaterializer<>((MaterializedInternal>) materializedInternal).materialize(); if (isGlobalKTable) { topologyBuilder.addGlobalStore( diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadOnlyDecorator.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadOnlyDecorator.java index 3ec8d7fa11508..561c5676417b9 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadOnlyDecorator.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadOnlyDecorator.java @@ -28,6 +28,8 @@ import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.TimestampedWindowStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; import org.apache.kafka.streams.state.WindowStore; import org.apache.kafka.streams.state.WindowStoreIterator; import org.apache.kafka.streams.state.internals.WrappedStateStore; @@ -68,6 +70,8 @@ public void close() { static StateStore getReadOnlyStore(final StateStore global) { if (global instanceof TimestampedKeyValueStore) { return new TimestampedKeyValueStoreReadOnlyDecorator<>((TimestampedKeyValueStore) global); + } else if (global instanceof VersionedKeyValueStore) { + return new VersionedKeyValueStoreReadOnlyDecorator<>((VersionedKeyValueStore) global); } else if (global instanceof KeyValueStore) { return new KeyValueStoreReadOnlyDecorator<>((KeyValueStore) global); } else if (global instanceof TimestampedWindowStore) { @@ -159,6 +163,35 @@ private TimestampedKeyValueStoreReadOnlyDecorator(final TimestampedKeyValueStore } } + static class VersionedKeyValueStoreReadOnlyDecorator + extends AbstractReadOnlyDecorator, K, V> + implements VersionedKeyValueStore { + + private VersionedKeyValueStoreReadOnlyDecorator(final VersionedKeyValueStore inner) { + super(inner); + } + + @Override + public void put(final K key, final V value, final long timestamp) { + throw new UnsupportedOperationException(ERROR_MESSAGE); + } + + @Override + public VersionedRecord delete(final K key, final long timestamp) { + throw new UnsupportedOperationException(ERROR_MESSAGE); + } + + @Override + public VersionedRecord get(final K key) { + return wrapped().get(key); + } + + @Override + public VersionedRecord get(final K key, final long asOfTimestamp) { + return wrapped().get(key, asOfTimestamp); + } + } + static class WindowStoreReadOnlyDecorator extends AbstractReadOnlyDecorator, K, V> implements WindowStore { diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java index 3c7f70ea0706e..1207039a552ae 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java @@ -28,6 +28,8 @@ import org.apache.kafka.streams.state.TimestampedKeyValueStore; import org.apache.kafka.streams.state.TimestampedWindowStore; import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; import org.apache.kafka.streams.state.WindowStore; import org.apache.kafka.streams.state.WindowStoreIterator; import org.apache.kafka.streams.state.internals.WrappedStateStore; @@ -62,6 +64,8 @@ public void close() { static StateStore getReadWriteStore(final StateStore store) { if (store instanceof TimestampedKeyValueStore) { return new TimestampedKeyValueStoreReadWriteDecorator<>((TimestampedKeyValueStore) store); + } else if (store instanceof VersionedKeyValueStore) { + return new VersionedKeyValueStoreReadWriteDecorator<>((VersionedKeyValueStore) store); } else if (store instanceof KeyValueStore) { return new KeyValueStoreReadWriteDecorator<>((KeyValueStore) store); } else if (store instanceof TimestampedWindowStore) { @@ -153,6 +157,35 @@ static class TimestampedKeyValueStoreReadWriteDecorator } } + static class VersionedKeyValueStoreReadWriteDecorator + extends AbstractReadWriteDecorator, K, V> + implements VersionedKeyValueStore { + + VersionedKeyValueStoreReadWriteDecorator(final VersionedKeyValueStore inner) { + super(inner); + } + + @Override + public void put(final K key, final V value, final long timestamp) { + wrapped().put(key, value, timestamp); + } + + @Override + public VersionedRecord delete(final K key, final long timestamp) { + return wrapped().delete(key, timestamp); + } + + @Override + public VersionedRecord get(final K key) { + return wrapped().get(key); + } + + @Override + public VersionedRecord get(final K key, final long asOfTimestamp) { + return wrapped().get(key, asOfTimestamp); + } + } + static class WindowStoreReadWriteDecorator extends AbstractReadWriteDecorator, K, V> implements WindowStore { diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilder.java new file mode 100644 index 0000000000000..ced78d49bf366 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilder.java @@ -0,0 +1,80 @@ +/* + * 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.common.serialization.Serde; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.util.Objects; +import org.apache.kafka.streams.state.StoreBuilder; +import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.apache.kafka.streams.state.VersionedBytesStore; +import org.apache.kafka.streams.state.VersionedBytesStoreSupplier; +import org.apache.kafka.streams.state.VersionedKeyValueStore; + +public class VersionedKeyValueStoreBuilder + extends AbstractStoreBuilder> { + + private final VersionedBytesStoreSupplier storeSupplier; + + public VersionedKeyValueStoreBuilder(final VersionedBytesStoreSupplier storeSupplier, + final Serde keySerde, + final Serde valueSerde, + final Time time) { + super( + storeSupplier.name(), + keySerde, + valueSerde, + time); + Objects.requireNonNull(storeSupplier, "storeSupplier can't be null"); + Objects.requireNonNull(storeSupplier.metricsScope(), "storeSupplier's metricsScope can't be null"); + this.storeSupplier = storeSupplier; + } + + @Override + public VersionedKeyValueStore build() { + final KeyValueStore store = storeSupplier.get(); + if (!(store instanceof VersionedBytesStore)) { + throw new IllegalStateException("VersionedBytesStoreSupplier.get() must return an instance of VersionedBytesStore"); + } + + final Serde> valueAndTimestampSerde = valueSerde == null + ? null + : new NullableValueAndTimestampSerde<>(valueSerde); + + return new MeteredVersionedKeyValueStore<>( + maybeWrapLogging((VersionedBytesStore) store), // no caching layer for versioned stores + storeSupplier.metricsScope(), + time, + keySerde, + valueAndTimestampSerde); + } + + @Override + public StoreBuilder> withCachingEnabled() { + throw new IllegalStateException("Versioned stores do not support caching"); + } + + private VersionedBytesStore maybeWrapLogging(final VersionedBytesStore inner) { + if (!enableLogging) { + return inner; + } + return new ChangeLoggingVersionedKeyValueBytesStore(inner); + } +} \ No newline at end of file diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java index 391fa09b07c38..f9ec855231870 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStreamThreadTest.java @@ -31,7 +31,7 @@ import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.internals.InternalNameProvider; import org.apache.kafka.streams.kstream.internals.MaterializedInternal; -import org.apache.kafka.streams.kstream.internals.TimestampedKeyValueStoreMaterializer; +import org.apache.kafka.streams.kstream.internals.KeyValueStoreMaterializer; import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.api.ContextualProcessor; import org.apache.kafka.streams.processor.api.ProcessorSupplier; @@ -100,7 +100,7 @@ public void process(final Record record) { }; builder.addGlobalStore( - new TimestampedKeyValueStoreMaterializer<>(materialized).materialize().withLoggingDisabled(), + new KeyValueStoreMaterializer<>(materialized).materialize().withLoggingDisabled(), "sourceName", null, null, diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/KeyValueStoreMaterializerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/KeyValueStoreMaterializerTest.java new file mode 100644 index 0000000000000..1cf4fb9d4a074 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/KeyValueStoreMaterializerTest.java @@ -0,0 +1,245 @@ +/* + * 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.processor.internals; + +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.internals.InternalNameProvider; +import org.apache.kafka.streams.kstream.internals.MaterializedInternal; +import org.apache.kafka.streams.kstream.internals.KeyValueStoreMaterializer; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; +import org.apache.kafka.streams.state.KeyValueStore; +import org.apache.kafka.streams.state.StoreBuilder; +import org.apache.kafka.streams.state.TimestampedKeyValueStore; +import org.apache.kafka.streams.state.VersionedBytesStore; +import org.apache.kafka.streams.state.VersionedBytesStoreSupplier; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.internals.CachingKeyValueStore; +import org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore; +import org.apache.kafka.streams.state.internals.ChangeLoggingTimestampedKeyValueBytesStore; +import org.apache.kafka.streams.state.internals.ChangeLoggingVersionedKeyValueBytesStore; +import org.apache.kafka.streams.state.internals.InMemoryKeyValueStore; +import org.apache.kafka.streams.state.internals.MeteredTimestampedKeyValueStore; +import org.apache.kafka.streams.state.internals.MeteredVersionedKeyValueStore; +import org.apache.kafka.streams.state.internals.WrappedStateStore; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.core.IsInstanceOf.instanceOf; +import static org.hamcrest.core.IsNot.not; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.StrictStubs.class) +public class KeyValueStoreMaterializerTest { + + private static final String STORE_PREFIX = "prefix"; + private static final String STORE_NAME = "name"; + private static final String METRICS_SCOPE = "metricScope"; + + @Mock + private InternalNameProvider nameProvider; + @Mock + private KeyValueBytesStoreSupplier keyValueStoreSupplier; + @Mock + private VersionedBytesStoreSupplier versionedStoreSupplier; + private final KeyValueStore innerKeyValueStore = new InMemoryKeyValueStore(STORE_NAME); + @Mock + private VersionedBytesStore innerVersionedStore; + + @Before + public void setUp() { + when(keyValueStoreSupplier.get()).thenReturn(innerKeyValueStore); + when(keyValueStoreSupplier.name()).thenReturn(STORE_NAME); + when(keyValueStoreSupplier.metricsScope()).thenReturn(METRICS_SCOPE); + + when(innerVersionedStore.name()).thenReturn(STORE_NAME); + when(versionedStoreSupplier.get()).thenReturn(innerVersionedStore); + when(versionedStoreSupplier.name()).thenReturn(STORE_NAME); + when(versionedStoreSupplier.metricsScope()).thenReturn(METRICS_SCOPE); + } + + @Test + public void shouldCreateTimestampedBuilderWithCachingAndLoggingEnabledByDefault() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as("store"), nameProvider, STORE_PREFIX); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + final StateStore logging = caching.wrapped(); + assertThat(store, instanceOf(MeteredTimestampedKeyValueStore.class)); + assertThat(caching, instanceOf(CachingKeyValueStore.class)); + assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStore.class)); + } + + @Test + public void shouldCreateDefaultTimestampedBuilderWithCachingDisabled() { + final MaterializedInternal> materialized = new MaterializedInternal<>( + Materialized.>as("store").withCachingDisabled(), nameProvider, STORE_PREFIX + ); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + assertThat(logging, instanceOf(ChangeLoggingKeyValueBytesStore.class)); + } + + @Test + public void shouldCreateDefaultTimestampedBuilderWithLoggingDisabled() { + final MaterializedInternal> materialized = new MaterializedInternal<>( + Materialized.>as("store").withLoggingDisabled(), nameProvider, STORE_PREFIX + ); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + assertThat(caching, instanceOf(CachingKeyValueStore.class)); + assertThat(caching.wrapped(), not(instanceOf(ChangeLoggingKeyValueBytesStore.class))); + } + + @Test + public void shouldCreateDefaultTimestampedBuilderWithCachingAndLoggingDisabled() { + final MaterializedInternal> materialized = new MaterializedInternal<>( + Materialized.>as("store").withCachingDisabled().withLoggingDisabled(), nameProvider, STORE_PREFIX + ); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final StateStore wrapped = ((WrappedStateStore) store).wrapped(); + assertThat(wrapped, not(instanceOf(CachingKeyValueStore.class))); + assertThat(wrapped, not(instanceOf(ChangeLoggingKeyValueBytesStore.class))); + } + + @Test + public void shouldCreateTimestampedStoreWithProvidedSupplierAndCachingAndLoggingEnabledByDefault() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(keyValueStoreSupplier), nameProvider, STORE_PREFIX); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + final StateStore logging = caching.wrapped(); + assertThat(innerKeyValueStore.name(), equalTo(store.name())); + assertThat(store, instanceOf(MeteredTimestampedKeyValueStore.class)); + assertThat(caching, instanceOf(CachingKeyValueStore.class)); + assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStore.class)); + } + + @Test + public void shouldCreateTimestampedStoreWithProvidedSupplierAndCachingDisabled() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(keyValueStoreSupplier).withCachingDisabled(), nameProvider, STORE_PREFIX); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + assertThat(innerKeyValueStore.name(), equalTo(store.name())); + assertThat(logging, instanceOf(ChangeLoggingKeyValueBytesStore.class)); + } + + @Test + public void shouldCreateTimestampedStoreWithProvidedSupplierAndLoggingDisabled() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(keyValueStoreSupplier).withLoggingDisabled(), nameProvider, STORE_PREFIX); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + assertThat(innerKeyValueStore.name(), equalTo(store.name())); + assertThat(caching, instanceOf(CachingKeyValueStore.class)); + assertThat(caching.wrapped(), not(instanceOf(ChangeLoggingKeyValueBytesStore.class))); + } + + @Test + public void shouldCreateTimestampedStoreWithProvidedSupplierAndCachingAndLoggingDisabled() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(keyValueStoreSupplier).withCachingDisabled().withLoggingDisabled(), nameProvider, STORE_PREFIX); + + final TimestampedKeyValueStore store = getTimestampedStore(materialized); + + final StateStore wrapped = ((WrappedStateStore) store).wrapped(); + assertThat(innerKeyValueStore.name(), equalTo(store.name())); + assertThat(wrapped, not(instanceOf(CachingKeyValueStore.class))); + assertThat(wrapped, not(instanceOf(ChangeLoggingKeyValueBytesStore.class))); + } + + @Test + public void shouldCreateVersionedStoreWithProvidedSupplierAndLoggingEnabledByDefault() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(versionedStoreSupplier), nameProvider, STORE_PREFIX); + + final VersionedKeyValueStore store = getVersionedStore(materialized); + + final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + final StateStore inner = logging.wrapped(); + assertThat(innerVersionedStore.name(), equalTo(store.name())); + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + assertThat(logging, instanceOf(ChangeLoggingVersionedKeyValueBytesStore.class)); + assertThat(innerVersionedStore, equalTo(inner)); + } + + @Test + public void shouldCreateVersionedStoreWithProvidedSupplierAndLoggingDisabled() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(versionedStoreSupplier).withLoggingDisabled(), nameProvider, STORE_PREFIX); + + final VersionedKeyValueStore store = getVersionedStore(materialized); + + final StateStore inner = ((WrappedStateStore) store).wrapped(); + assertThat(innerVersionedStore.name(), equalTo(store.name())); + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + assertThat(innerVersionedStore, equalTo(inner)); + } + + @Test + public void shouldNotBuildVersionedStoreWithCachingEvenIfExplicitlySet() { + final MaterializedInternal> materialized = + new MaterializedInternal<>(Materialized.as(versionedStoreSupplier).withCachingEnabled(), nameProvider, STORE_PREFIX); + + final VersionedKeyValueStore store = getVersionedStore(materialized); + + final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); + final StateStore inner = logging.wrapped(); + assertThat(innerVersionedStore.name(), equalTo(store.name())); + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + assertThat(logging, instanceOf(ChangeLoggingVersionedKeyValueBytesStore.class)); + assertThat(innerVersionedStore, equalTo(inner)); + } + + @SuppressWarnings("unchecked") + private static TimestampedKeyValueStore getTimestampedStore( + final MaterializedInternal> materialized) { + final KeyValueStoreMaterializer materializer = new KeyValueStoreMaterializer<>(materialized); + final StoreBuilder builder = materializer.materialize(); + return (TimestampedKeyValueStore) builder.build(); + } + + @SuppressWarnings("unchecked") + private static VersionedKeyValueStore getVersionedStore( + final MaterializedInternal> materialized) { + final KeyValueStoreMaterializer materializer = new KeyValueStoreMaterializer<>(materialized); + final StoreBuilder builder = materializer.materialize(); + return (VersionedKeyValueStore) builder.build(); + } +} \ No newline at end of file diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TimestampedKeyValueStoreMaterializerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TimestampedKeyValueStoreMaterializerTest.java deleted file mode 100644 index cb67636f69354..0000000000000 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TimestampedKeyValueStoreMaterializerTest.java +++ /dev/null @@ -1,125 +0,0 @@ -/* - * 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.processor.internals; - -import org.apache.kafka.common.utils.Bytes; -import org.apache.kafka.streams.kstream.Materialized; -import org.apache.kafka.streams.kstream.internals.InternalNameProvider; -import org.apache.kafka.streams.kstream.internals.MaterializedInternal; -import org.apache.kafka.streams.kstream.internals.TimestampedKeyValueStoreMaterializer; -import org.apache.kafka.streams.processor.StateStore; -import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; -import org.apache.kafka.streams.state.KeyValueStore; -import org.apache.kafka.streams.state.StoreBuilder; -import org.apache.kafka.streams.state.TimestampedKeyValueStore; -import org.apache.kafka.streams.state.internals.CachingKeyValueStore; -import org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore; -import org.apache.kafka.streams.state.internals.ChangeLoggingTimestampedKeyValueBytesStore; -import org.apache.kafka.streams.state.internals.InMemoryKeyValueStore; -import org.apache.kafka.streams.state.internals.MeteredTimestampedKeyValueStore; -import org.apache.kafka.streams.state.internals.WrappedStateStore; -import org.hamcrest.CoreMatchers; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mock; -import org.mockito.junit.MockitoJUnitRunner; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.core.IsInstanceOf.instanceOf; -import static org.hamcrest.core.IsNot.not; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -@RunWith(MockitoJUnitRunner.StrictStubs.class) -public class TimestampedKeyValueStoreMaterializerTest { - - private final String storePrefix = "prefix"; - @Mock - private InternalNameProvider nameProvider; - - @Test - public void shouldCreateBuilderThatBuildsMeteredStoreWithCachingAndLoggingEnabled() { - final MaterializedInternal> materialized = - new MaterializedInternal<>(Materialized.as("store"), nameProvider, storePrefix); - - final TimestampedKeyValueStoreMaterializer materializer = new TimestampedKeyValueStoreMaterializer<>(materialized); - final StoreBuilder> builder = materializer.materialize(); - final TimestampedKeyValueStore store = builder.build(); - final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); - final StateStore logging = caching.wrapped(); - assertThat(store, instanceOf(MeteredTimestampedKeyValueStore.class)); - assertThat(caching, instanceOf(CachingKeyValueStore.class)); - assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStore.class)); - } - - @Test - public void shouldCreateBuilderThatBuildsStoreWithCachingDisabled() { - final MaterializedInternal> materialized = new MaterializedInternal<>( - Materialized.>as("store").withCachingDisabled(), nameProvider, storePrefix - ); - final TimestampedKeyValueStoreMaterializer materializer = new TimestampedKeyValueStoreMaterializer<>(materialized); - final StoreBuilder> builder = materializer.materialize(); - final TimestampedKeyValueStore store = builder.build(); - final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); - assertThat(logging, instanceOf(ChangeLoggingKeyValueBytesStore.class)); - } - - @Test - public void shouldCreateBuilderThatBuildsStoreWithLoggingDisabled() { - final MaterializedInternal> materialized = new MaterializedInternal<>( - Materialized.>as("store").withLoggingDisabled(), nameProvider, storePrefix - ); - final TimestampedKeyValueStoreMaterializer materializer = new TimestampedKeyValueStoreMaterializer<>(materialized); - final StoreBuilder> builder = materializer.materialize(); - final TimestampedKeyValueStore store = builder.build(); - final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); - assertThat(caching, instanceOf(CachingKeyValueStore.class)); - assertThat(caching.wrapped(), not(instanceOf(ChangeLoggingKeyValueBytesStore.class))); - } - - @Test - public void shouldCreateBuilderThatBuildsStoreWithCachingAndLoggingDisabled() { - final MaterializedInternal> materialized = new MaterializedInternal<>( - Materialized.>as("store").withCachingDisabled().withLoggingDisabled(), nameProvider, storePrefix - ); - final TimestampedKeyValueStoreMaterializer materializer = new TimestampedKeyValueStoreMaterializer<>(materialized); - final StoreBuilder> builder = materializer.materialize(); - final TimestampedKeyValueStore store = builder.build(); - final StateStore wrapped = ((WrappedStateStore) store).wrapped(); - assertThat(wrapped, not(instanceOf(CachingKeyValueStore.class))); - assertThat(wrapped, not(instanceOf(ChangeLoggingKeyValueBytesStore.class))); - } - - @Test - public void shouldCreateKeyValueStoreWithTheProvidedInnerStore() { - final KeyValueBytesStoreSupplier supplier = mock(KeyValueBytesStoreSupplier.class); - final InMemoryKeyValueStore store = new InMemoryKeyValueStore("name"); - when(supplier.name()).thenReturn("name"); - when(supplier.get()).thenReturn(store); - when(supplier.metricsScope()).thenReturn("metricScope"); - - final MaterializedInternal> materialized = - new MaterializedInternal<>(Materialized.as(supplier), nameProvider, storePrefix); - final TimestampedKeyValueStoreMaterializer materializer = new TimestampedKeyValueStoreMaterializer<>(materialized); - final StoreBuilder> builder = materializer.materialize(); - final TimestampedKeyValueStore built = builder.build(); - - assertThat(store.name(), CoreMatchers.equalTo(built.name())); - } - -} \ No newline at end of file diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderTest.java index b79d67e030924..db02483359149 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderTest.java @@ -97,8 +97,8 @@ public void shouldHaveCachingStoreWhenEnabled() { @Test public void shouldHaveChangeLoggingStoreWhenLoggingEnabled() { final TimestampedKeyValueStore store = builder - .withLoggingEnabled(Collections.emptyMap()) - .build(); + .withLoggingEnabled(Collections.emptyMap()) + .build(); final StateStore wrapped = ((WrappedStateStore) store).wrapped(); assertThat(store, instanceOf(MeteredTimestampedKeyValueStore.class)); assertThat(wrapped, instanceOf(ChangeLoggingTimestampedKeyValueBytesStore.class)); @@ -108,9 +108,9 @@ public void shouldHaveChangeLoggingStoreWhenLoggingEnabled() { @Test public void shouldHaveCachingAndChangeLoggingWhenBothEnabled() { final TimestampedKeyValueStore store = builder - .withLoggingEnabled(Collections.emptyMap()) - .withCachingEnabled() - .build(); + .withLoggingEnabled(Collections.emptyMap()) + .withCachingEnabled() + .build(); final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrapped(); final WrappedStateStore changeLogging = (WrappedStateStore) caching.wrapped(); assertThat(store, instanceOf(MeteredTimestampedKeyValueStore.class)); @@ -154,17 +154,34 @@ public void shouldThrowNullPointerIfInnerIsNull() { } @Test - public void shouldThrowNullPointerIfKeySerdeIsNull() { - assertThrows(NullPointerException.class, () -> new TimestampedKeyValueStoreBuilder<>(supplier, null, Serdes.String(), new MockTime())); + public void shouldNotThrowNullPointerIfKeySerdeIsNull() { + reset(supplier); + expect(supplier.name()).andReturn("name"); + expect(supplier.metricsScope()).andReturn("metricScope").anyTimes(); + replay(supplier); + + // does not throw + new TimestampedKeyValueStoreBuilder<>(supplier, null, Serdes.String(), new MockTime()); } @Test - public void shouldThrowNullPointerIfValueSerdeIsNull() { - assertThrows(NullPointerException.class, () -> new TimestampedKeyValueStoreBuilder<>(supplier, Serdes.String(), null, new MockTime())); + public void shouldNotThrowNullPointerIfValueSerdeIsNull() { + reset(supplier); + expect(supplier.name()).andReturn("name"); + expect(supplier.metricsScope()).andReturn("metricScope").anyTimes(); + replay(supplier); + + // does not throw + new TimestampedKeyValueStoreBuilder<>(supplier, Serdes.String(), null, new MockTime()); } @Test public void shouldThrowNullPointerIfTimeIsNull() { + reset(supplier); + expect(supplier.name()).andReturn("name"); + expect(supplier.metricsScope()).andReturn("metricScope").anyTimes(); + replay(supplier); + assertThrows(NullPointerException.class, () -> new TimestampedKeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), null)); } diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilderTest.java new file mode 100644 index 0000000000000..49ab923c6a355 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/VersionedKeyValueStoreBuilderTest.java @@ -0,0 +1,146 @@ +/* + * 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.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.core.IsInstanceOf.instanceOf; +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.when; + +import java.util.Collections; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.state.VersionedBytesStore; +import org.apache.kafka.streams.state.VersionedBytesStoreSupplier; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.StrictStubs.class) +public class VersionedKeyValueStoreBuilderTest { + + private static final String STORE_NAME = "versioned-store"; + private static final String METRICS_SCOPE = "metrics-scope"; + + @Mock + private VersionedBytesStoreSupplier supplier; + @Mock + private VersionedBytesStore inner; + + private VersionedKeyValueStoreBuilder builder; + + @Before + public void setUp() { + when(supplier.get()).thenReturn(inner); + when(supplier.name()).thenReturn(STORE_NAME); + when(supplier.metricsScope()).thenReturn(METRICS_SCOPE); + + builder = new VersionedKeyValueStoreBuilder<>( + supplier, + Serdes.String(), + Serdes.String(), + new MockTime() + ); + } + + @Test + public void shouldHaveMeteredStoreAsOuterStore() { + final VersionedKeyValueStore store = builder.build(); + + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + } + + @Test + public void shouldHaveChangeLoggingStoreByDefault() { + final VersionedKeyValueStore store = builder.build(); + + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + final StateStore next = ((WrappedStateStore) store).wrapped(); + assertThat(next, instanceOf(ChangeLoggingVersionedKeyValueBytesStore.class)); + } + + @Test + public void shouldNotHaveChangeLoggingStoreWhenDisabled() { + final VersionedKeyValueStore store = builder + .withLoggingDisabled() + .build(); + + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + final StateStore next = ((WrappedStateStore) store).wrapped(); + assertThat(next, equalTo(inner)); + + } + + @Test + public void shouldHaveChangeLoggingStoreWhenLoggingEnabled() { + final VersionedKeyValueStore store = builder + .withLoggingEnabled(Collections.emptyMap()) + .build(); + + assertThat(store, instanceOf(MeteredVersionedKeyValueStore.class)); + final StateStore next = ((WrappedStateStore) store).wrapped(); + assertThat(next, instanceOf(ChangeLoggingVersionedKeyValueBytesStore.class)); + assertThat(((WrappedStateStore) next).wrapped(), equalTo(inner)); + } + + @Test + public void shouldThrowWhenCachingEnabled() { + assertThrows(IllegalStateException.class, () -> builder.withCachingEnabled()); + } + + @SuppressWarnings("all") + @Test + public void shouldThrowNullPointerIfInnerIsNull() { + assertThrows(NullPointerException.class, () -> new VersionedKeyValueStoreBuilder<>(null, Serdes.String(), Serdes.String(), new MockTime())); + } + + @Test + public void shouldNotThrowNullPointerIfKeySerdeIsNull() { + // does not throw + new VersionedKeyValueStoreBuilder<>(supplier, null, Serdes.String(), new MockTime()); + } + + @Test + public void shouldNotThrowNullPointerIfValueSerdeIsNull() { + // does not throw + new VersionedKeyValueStoreBuilder<>(supplier, Serdes.String(), null, new MockTime()); + } + + @Test + public void shouldThrowNullPointerIfTimeIsNull() { + assertThrows(NullPointerException.class, () -> new VersionedKeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), null)); + } + + @Test + public void shouldThrowNullPointerIfNameIsNull() { + when(supplier.name()).thenReturn(null); + + assertThrows(NullPointerException.class, () -> new VersionedKeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), new MockTime())); + } + + @Test + public void shouldThrowNullPointerIfMetricsScopeIsNull() { + when(supplier.metricsScope()).thenReturn(null); + + assertThrows(NullPointerException.class, () -> new VersionedKeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), new MockTime())); + } +} \ No newline at end of file From 5edfc3dc11f4b7379ba3e2ad3a779249ec4eb262 Mon Sep 17 00:00:00 2001 From: Victoria Xia Date: Fri, 3 Mar 2023 00:15:01 -0800 Subject: [PATCH 2/2] review feedback --- .../internals/KeyValueStoreMaterializer.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java index 5f8440894e1bd..33792dafb713d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java @@ -25,12 +25,16 @@ import org.apache.kafka.streams.state.VersionedBytesStoreSupplier; import org.apache.kafka.streams.state.internals.TimestampedKeyValueStoreBuilder; import org.apache.kafka.streams.state.internals.VersionedKeyValueStoreBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Materializes a key-value store as either a {@link TimestampedKeyValueStoreBuilder} or a * {@link VersionedKeyValueStoreBuilder} depending on whether the store is versioned or not. */ public class KeyValueStoreMaterializer { + private static final Logger LOG = LoggerFactory.getLogger(KeyValueStoreMaterializer.class); + private final MaterializedInternal> materialized; public KeyValueStoreMaterializer(final MaterializedInternal> materialized) { @@ -75,9 +79,12 @@ public StoreBuilder materialize() { builder.withLoggingDisabled(); } - // versioned stores do not support caching - if (materialized.cachingEnabled() && !(builder instanceof VersionedKeyValueStoreBuilder)) { - builder.withCachingEnabled(); + if (materialized.cachingEnabled()) { + if (!(builder instanceof VersionedKeyValueStoreBuilder)) { + builder.withCachingEnabled(); + } else { + LOG.info("Not enabling caching for store '{}' as versioned stores do not support caching.", supplier.name()); + } } return builder; }