This is an automated email from the ASF dual-hosted git repository. mjsax pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push: new b6c1cb0eec4 MINOR: update CachingPersistentWindowStoreTest (#16701) b6c1cb0eec4 is described below commit b6c1cb0eec4cad56f8db6eba05d24f6b3c44a211 Author: Matthias J. Sax <matth...@confluent.io> AuthorDate: Mon Jul 29 12:45:13 2024 -0700 MINOR: update CachingPersistentWindowStoreTest (#16701) Refactor test to move off deprecated `transform()` in favor of `process()`. Reviewers: Bill Bejeck <b...@confluent.io> --- .../CachingPersistentWindowStoreTest.java | 41 ++++++++++------------ 1 file changed, 18 insertions(+), 23 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingPersistentWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingPersistentWindowStoreTest.java index 07293d2d4b9..6a055f51cf9 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingPersistentWindowStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingPersistentWindowStoreTest.java @@ -32,11 +32,12 @@ import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.errors.InvalidStateStoreException; import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.TimeWindowedDeserializer; -import org.apache.kafka.streams.kstream.Transformer; import org.apache.kafka.streams.kstream.Windowed; import org.apache.kafka.streams.kstream.internals.TimeWindow; -import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.StateStoreContext; +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.processor.internals.MockStreamsMetrics; import org.apache.kafka.streams.processor.internals.ProcessorRecordContext; import org.apache.kafka.streams.query.Position; @@ -102,7 +103,7 @@ public class CachingPersistentWindowStoreTest { private static final String TOPIC = "topic"; private static final String CACHE_NAMESPACE = "0_0-store-name"; - private InternalMockProcessorContext context; + private InternalMockProcessorContext<?, ?> context; private RocksDBSegmentedBytesStore bytesStore; private WindowStore<Bytes, byte[]> underlyingStore; private CachingWindowStore cachingStore; @@ -138,8 +139,8 @@ public class CachingPersistentWindowStoreTest { final WindowStore<Bytes, byte[]> inner = mock(WindowStore.class); final CachingWindowStore outer = new CachingWindowStore(inner, WINDOW_SIZE, SEGMENT_INTERVAL); when(inner.name()).thenReturn("store"); - outer.init((ProcessorContext) context, outer); - verify(inner).init((ProcessorContext) context, outer); + outer.init((org.apache.kafka.streams.processor.ProcessorContext) context, outer); + verify(inner).init((org.apache.kafka.streams.processor.ProcessorContext) context, outer); } @SuppressWarnings("unchecked") @@ -153,30 +154,28 @@ public class CachingPersistentWindowStoreTest { } @Test - @SuppressWarnings("deprecation") public void shouldNotReturnDuplicatesInRanges() { final StreamsBuilder builder = new StreamsBuilder(); final StoreBuilder<WindowStore<String, String>> storeBuilder = Stores.windowStoreBuilder( - Stores.persistentWindowStore("store-name", ofHours(1L), ofMinutes(1L), false), - Serdes.String(), - Serdes.String()) + Stores.persistentWindowStore("store-name", ofHours(1L), ofMinutes(1L), false), + Serdes.String(), + Serdes.String()) .withCachingEnabled(); builder.addStateStore(storeBuilder); builder.stream(TOPIC, Consumed.with(Serdes.String(), Serdes.String())) - .transform(() -> new Transformer<String, String, KeyValue<String, String>>() { + .process(() -> new Processor<String, String, String, String>() { private WindowStore<String, String> store; private int numRecordsProcessed; - private ProcessorContext context; + private ProcessorContext<String, String> context; - @SuppressWarnings("unchecked") @Override - public void init(final ProcessorContext processorContext) { + public void init(final ProcessorContext<String, String> processorContext) { this.context = processorContext; - this.store = (WindowStore<String, String>) processorContext.getStateStore("store-name"); + this.store = processorContext.getStateStore("store-name"); int count = 0; try (final KeyValueIterator<Windowed<String>, String> all = store.all()) { @@ -190,7 +189,7 @@ public class CachingPersistentWindowStoreTest { } @Override - public KeyValue<String, String> transform(final String key, final String value) { + public void process(final Record<String, String> record) { int count = 0; try (final KeyValueIterator<Windowed<String>, String> all = store.all()) { @@ -202,22 +201,18 @@ public class CachingPersistentWindowStoreTest { assertThat(count, equalTo(numRecordsProcessed)); - store.put(value, value, context.timestamp()); + store.put(record.value(), record.value(), record.timestamp()); numRecordsProcessed++; - return new KeyValue<>(key, value); - } - - @Override - public void close() { + context.forward(record); } }, "store-name"); final Properties streamsConfiguration = new Properties(); streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath()); streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000L);