kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mj...@apache.org
Subject [kafka] branch trunk updated: KAFAK-3522: Add TopologyTestDriver unit tests (#6179)
Date Thu, 16 May 2019 15:16:23 GMT
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 16b4088  KAFAK-3522: Add TopologyTestDriver unit tests (#6179)
16b4088 is described below

commit 16b408898e75b00ddf6b607246833cdbcd56f507
Author: Matthias J. Sax <matthias@confluent.io>
AuthorDate: Thu May 16 16:16:00 2019 +0100

    KAFAK-3522: Add TopologyTestDriver unit tests (#6179)
    
    Reviewers: John Roesler <john@confluent.io>, Bill Bejeck <bill@confluent.io>
---
 .../org/apache/kafka/streams/state/Stores.java     |  41 +-
 .../integration/StoreUpgradeIntegrationTest.java   | 995 +++++++++++++++++++++
 .../kafka/streams/TopologyTestDriverTest.java      | 196 +++-
 3 files changed, 1188 insertions(+), 44 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
index e40251d..2f81fd3 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
@@ -354,26 +354,6 @@ public final class Stores {
     }
 
     /**
-     * Create an in-memory {@link SessionBytesStoreSupplier}.
-     * @param name              name of the store (cannot be {@code null})
-     * @param retentionPeriod   length ot time to retain data in the store (cannot be negative)
-     *                          Note that the retention period must be at least long enough to contain the
-     *                          windowed data's entire life cycle, from window-start through window-end,
-     *                          and for the entire grace period.
-     * @return an instance of a {@link  SessionBytesStoreSupplier}
-     */
-    public static SessionBytesStoreSupplier inMemorySessionStore(final String name, final Duration retentionPeriod) {
-        Objects.requireNonNull(name, "name cannot be null");
-
-        final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod");
-        final long retentionPeriodMs = ApiUtils.validateMillisecondDuration(retentionPeriod, msgPrefix);
-        if (retentionPeriodMs < 0) {
-            throw new IllegalArgumentException("retentionPeriod cannot be negative");
-        }
-        return new InMemorySessionBytesStoreSupplier(name, retentionPeriodMs);
-    }
-
-    /**
      * Create a persistent {@link SessionBytesStoreSupplier}.
      *
      * @param name              name of the store (cannot be {@code null})
@@ -412,6 +392,27 @@ public final class Stores {
     }
 
     /**
+     * Create an in-memory {@link SessionBytesStoreSupplier}.
+     *
+     * @param name              name of the store (cannot be {@code null})
+     * @param retentionPeriod   length ot time to retain data in the store (cannot be negative)
+     *                          Note that the retention period must be at least long enough to contain the
+     *                          windowed data's entire life cycle, from window-start through window-end,
+     *                          and for the entire grace period.
+     * @return an instance of a {@link  SessionBytesStoreSupplier}
+     */
+    public static SessionBytesStoreSupplier inMemorySessionStore(final String name, final Duration retentionPeriod) {
+        Objects.requireNonNull(name, "name cannot be null");
+
+        final String msgPrefix = prepareMillisCheckFailMsgPrefix(retentionPeriod, "retentionPeriod");
+        final long retentionPeriodMs = ApiUtils.validateMillisecondDuration(retentionPeriod, msgPrefix);
+        if (retentionPeriodMs < 0) {
+            throw new IllegalArgumentException("retentionPeriod cannot be negative");
+        }
+        return new InMemorySessionBytesStoreSupplier(name, retentionPeriodMs);
+    }
+
+    /**
      * Creates a {@link StoreBuilder} that can be used to build a {@link KeyValueStore}.
      * <p>
      * The provided supplier should <strong>not</strong> be a supplier for
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StoreUpgradeIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StoreUpgradeIntegrationTest.java
new file mode 100644
index 0000000..ba5b08f
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/StoreUpgradeIntegrationTest.java
@@ -0,0 +1,995 @@
+/*
+ * 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.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.QueryableStoreTypes;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyWindowStore;
+import org.apache.kafka.streams.state.Stores;
+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.WindowStore;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.time.Duration;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
+
+@Category({IntegrationTest.class})
+public class StoreUpgradeIntegrationTest {
+    private static String inputStream;
+    private static final String STORE_NAME = "store";
+
+    private KafkaStreams kafkaStreams;
+    private static int testCounter = 0;
+
+    @ClassRule
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
+
+    @Before
+    public void createTopics() throws Exception {
+        inputStream = "input-stream-" + testCounter;
+        CLUSTER.createTopic(inputStream);
+    }
+
+    private Properties props() {
+        final Properties streamsConfiguration = new Properties();
+        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "addId-" + testCounter++);
+        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+        streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
+        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
+        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
+        streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
+        streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        return streamsConfiguration;
+    }
+
+    @After
+    public void shutdown() {
+        if (kafkaStreams != null) {
+            kafkaStreams.close(Duration.ofSeconds(30L));
+            kafkaStreams.cleanUp();
+        }
+    }
+
+    @Test
+    public void shouldMigrateInMemoryKeyValueStoreToTimestampedKeyValueStoreUsingPapi() throws Exception {
+        shouldMigrateKeyValueStoreToTimestampedKeyValueStoreUsingPapi(false);
+    }
+
+    @Test
+    public void shouldMigratePersistentKeyValueStoreToTimestampedKeyValueStoreUsingPapi() throws Exception {
+        shouldMigrateKeyValueStoreToTimestampedKeyValueStoreUsingPapi(true);
+    }
+
+    private void shouldMigrateKeyValueStoreToTimestampedKeyValueStoreUsingPapi(final boolean persistentStore) throws Exception {
+        final StreamsBuilder streamsBuilderForOldStore = new StreamsBuilder();
+
+        streamsBuilderForOldStore.addStateStore(
+            Stores.keyValueStoreBuilder(
+                persistentStore ? Stores.persistentKeyValueStore(STORE_NAME) : Stores.inMemoryKeyValueStore(STORE_NAME),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(KeyValueProcessor::new, STORE_NAME);
+
+        final Properties props = props();
+        kafkaStreams = new KafkaStreams(streamsBuilderForOldStore.build(), props);
+        kafkaStreams.start();
+
+        processKeyValueAndVerifyPlainCount(1, singletonList(KeyValue.pair(1, 1L)));
+
+        processKeyValueAndVerifyPlainCount(1, singletonList(KeyValue.pair(1, 2L)));
+        final long lastUpdateKeyOne = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        processKeyValueAndVerifyPlainCount(2, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L)));
+        final long lastUpdateKeyTwo = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        processKeyValueAndVerifyPlainCount(3, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L)));
+        final long lastUpdateKeyThree = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        processKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L),
+            KeyValue.pair(4, 1L)));
+
+        processKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L),
+            KeyValue.pair(4, 2L)));
+
+        processKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L),
+            KeyValue.pair(4, 3L)));
+        final long lastUpdateKeyFour = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        kafkaStreams.close();
+        kafkaStreams = null;
+
+
+
+        final StreamsBuilder streamsBuilderForNewStore = new StreamsBuilder();
+
+        streamsBuilderForNewStore.addStateStore(
+            Stores.timestampedKeyValueStoreBuilder(
+                persistentStore ? Stores.persistentTimestampedKeyValueStore(STORE_NAME) : Stores.inMemoryKeyValueStore(STORE_NAME),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(TimestampedKeyValueProcessor::new, STORE_NAME);
+
+        kafkaStreams = new KafkaStreams(streamsBuilderForNewStore.build(), props);
+        kafkaStreams.start();
+
+        verifyCountWithTimestamp(1, 2L, lastUpdateKeyOne);
+        verifyCountWithTimestamp(2, 1L, lastUpdateKeyTwo);
+        verifyCountWithTimestamp(3, 1L, lastUpdateKeyThree);
+        verifyCountWithTimestamp(4, 3L, lastUpdateKeyFour);
+
+        final long currentTime = CLUSTER.time.milliseconds();
+        processKeyValueAndVerifyCountWithTimestamp(1, currentTime + 42L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(1L, lastUpdateKeyTwo)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(4, ValueAndTimestamp.make(3L, lastUpdateKeyFour))));
+
+        processKeyValueAndVerifyCountWithTimestamp(2, currentTime + 45L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(4, ValueAndTimestamp.make(3L, lastUpdateKeyFour))));
+
+        // can process "out of order" record for different key
+        processKeyValueAndVerifyCountWithTimestamp(4, currentTime + 21L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(4, ValueAndTimestamp.make(4L, currentTime + 21L))));
+
+        processKeyValueAndVerifyCountWithTimestamp(4, currentTime + 42L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(4, ValueAndTimestamp.make(5L, currentTime + 42L))));
+
+        // out of order (same key) record should not reduce result timestamp
+        processKeyValueAndVerifyCountWithTimestamp(4, currentTime + 10L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(4, ValueAndTimestamp.make(6L, currentTime + 42L))));
+
+        kafkaStreams.close();
+    }
+
+    @Test
+    public void shouldProxyKeyValueStoreToTimestampedKeyValueStoreUsingPapi() throws Exception {
+        final StreamsBuilder streamsBuilderForOldStore = new StreamsBuilder();
+
+        streamsBuilderForOldStore.addStateStore(
+            Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore(STORE_NAME),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(KeyValueProcessor::new, STORE_NAME);
+
+        final Properties props = props();
+        kafkaStreams = new KafkaStreams(streamsBuilderForOldStore.build(), props);
+        kafkaStreams.start();
+
+        processKeyValueAndVerifyPlainCount(1, singletonList(KeyValue.pair(1, 1L)));
+
+        processKeyValueAndVerifyPlainCount(1, singletonList(KeyValue.pair(1, 2L)));
+
+        processKeyValueAndVerifyPlainCount(2, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L)));
+
+        processKeyValueAndVerifyPlainCount(3, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L)));
+
+        processKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L),
+            KeyValue.pair(4, 1L)));
+
+        processKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L),
+            KeyValue.pair(4, 2L)));
+
+        processKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(1, 2L),
+            KeyValue.pair(2, 1L),
+            KeyValue.pair(3, 1L),
+            KeyValue.pair(4, 3L)));
+
+        kafkaStreams.close();
+        kafkaStreams = null;
+
+
+
+        final StreamsBuilder streamsBuilderForNewStore = new StreamsBuilder();
+
+        streamsBuilderForNewStore.addStateStore(
+            Stores.timestampedKeyValueStoreBuilder(
+                Stores.persistentKeyValueStore(STORE_NAME),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(TimestampedKeyValueProcessor::new, STORE_NAME);
+
+        kafkaStreams = new KafkaStreams(streamsBuilderForNewStore.build(), props);
+        kafkaStreams.start();
+
+        verifyCountWithSurrogateTimestamp(1, 2L);
+        verifyCountWithSurrogateTimestamp(2, 1L);
+        verifyCountWithSurrogateTimestamp(3, 1L);
+        verifyCountWithSurrogateTimestamp(4, 3L);
+
+        processKeyValueAndVerifyCount(1, 42L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(4, ValueAndTimestamp.make(3L, -1L))));
+
+        processKeyValueAndVerifyCount(2, 45L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(4, ValueAndTimestamp.make(3L, -1L))));
+
+        // can process "out of order" record for different key
+        processKeyValueAndVerifyCount(4, 21L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(4, ValueAndTimestamp.make(4L, -1L))));
+
+        processKeyValueAndVerifyCount(4, 42L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(4, ValueAndTimestamp.make(5L, -1L))));
+
+        // out of order (same key) record should not reduce result timestamp
+        processKeyValueAndVerifyCount(4, 10L, asList(
+            KeyValue.pair(1, ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(2, ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(3, ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(4, ValueAndTimestamp.make(6L, -1L))));
+
+        kafkaStreams.close();
+    }
+
+    private <K, V> void processKeyValueAndVerifyPlainCount(final K key,
+                                                           final List<KeyValue<Integer, Object>> expectedStoreContent)
+            throws Exception {
+
+        IntegrationTestUtils.produceKeyValuesSynchronously(
+            inputStream,
+            singletonList(KeyValue.pair(key, 0)),
+            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+                IntegerSerializer.class,
+                IntegerSerializer.class),
+            CLUSTER.time);
+
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyKeyValueStore<K, V> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.keyValueStore());
+                try (final KeyValueIterator<K, V> all = store.all()) {
+                    final List<KeyValue<K, V>> storeContent = new LinkedList<>();
+                    while (all.hasNext()) {
+                        storeContent.add(all.next());
+                    }
+                    return storeContent.equals(expectedStoreContent);
+                }
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K> void verifyCountWithTimestamp(final K key,
+                                              final long value,
+                                              final long timestamp) throws Exception {
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyKeyValueStore<K, ValueAndTimestamp<Long>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedKeyValueStore());
+                final ValueAndTimestamp<Long> count = store.get(key);
+                return count.value() == value && count.timestamp() == timestamp;
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K> void verifyCountWithSurrogateTimestamp(final K key,
+                                                       final long value) throws Exception {
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyKeyValueStore<K, ValueAndTimestamp<Long>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedKeyValueStore());
+                final ValueAndTimestamp<Long> count = store.get(key);
+                return count.value() == value && count.timestamp() == -1L;
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K, V> void processKeyValueAndVerifyCount(final K key,
+                                                      final long timestamp,
+                                                      final List<KeyValue<Integer, Object>> expectedStoreContent)
+            throws Exception {
+
+        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+            inputStream,
+            singletonList(KeyValue.pair(key, 0)),
+            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+                IntegerSerializer.class,
+                IntegerSerializer.class),
+            timestamp);
+
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedKeyValueStore());
+                try (final KeyValueIterator<K, ValueAndTimestamp<V>> all = store.all()) {
+                    final List<KeyValue<K, ValueAndTimestamp<V>>> storeContent = new LinkedList<>();
+                    while (all.hasNext()) {
+                        storeContent.add(all.next());
+                    }
+                    return storeContent.equals(expectedStoreContent);
+                }
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K, V> void processKeyValueAndVerifyCountWithTimestamp(final K key,
+                                                                   final long timestamp,
+                                                                   final List<KeyValue<Integer, Object>> expectedStoreContent)
+        throws Exception {
+
+        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+            inputStream,
+            singletonList(KeyValue.pair(key, 0)),
+            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+                IntegerSerializer.class,
+                IntegerSerializer.class),
+            timestamp);
+
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedKeyValueStore());
+                try (final KeyValueIterator<K, ValueAndTimestamp<V>> all = store.all()) {
+                    final List<KeyValue<K, ValueAndTimestamp<V>>> storeContent = new LinkedList<>();
+                    while (all.hasNext()) {
+                        storeContent.add(all.next());
+                    }
+                    return storeContent.equals(expectedStoreContent);
+                }
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    @Test
+    public void shouldMigrateInMemoryWindowStoreToTimestampedWindowStoreUsingPapi() throws Exception {
+        final StreamsBuilder streamsBuilderForOldStore = new StreamsBuilder();
+        streamsBuilderForOldStore
+            .addStateStore(
+                Stores.windowStoreBuilder(
+                    Stores.inMemoryWindowStore(
+                        STORE_NAME,
+                        Duration.ofMillis(1000L),
+                        Duration.ofMillis(1000L),
+                        false),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(WindowedProcessor::new, STORE_NAME);
+
+        final StreamsBuilder streamsBuilderForNewStore = new StreamsBuilder();
+        streamsBuilderForNewStore
+            .addStateStore(
+                Stores.timestampedWindowStoreBuilder(
+                    Stores.inMemoryWindowStore(
+                        STORE_NAME,
+                        Duration.ofMillis(1000L),
+                        Duration.ofMillis(1000L),
+                        false),
+            Serdes.Integer(),
+            Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(TimestampedWindowedProcessor::new, STORE_NAME);
+
+
+        shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(
+            new KafkaStreams(streamsBuilderForOldStore.build(), props()),
+            new KafkaStreams(streamsBuilderForNewStore.build(), props()),
+            false);
+    }
+
+    @Test
+    public void shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi() throws Exception {
+        final StreamsBuilder streamsBuilderForOldStore = new StreamsBuilder();
+
+        streamsBuilderForOldStore
+            .addStateStore(
+                Stores.windowStoreBuilder(
+                    Stores.persistentWindowStore(
+                        STORE_NAME,
+                        Duration.ofMillis(1000L),
+                        Duration.ofMillis(1000L),
+                        false),
+                    Serdes.Integer(),
+                    Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(WindowedProcessor::new, STORE_NAME);
+
+        final StreamsBuilder streamsBuilderForNewStore = new StreamsBuilder();
+        streamsBuilderForNewStore
+            .addStateStore(
+                Stores.timestampedWindowStoreBuilder(
+                    Stores.persistentTimestampedWindowStore(
+                        STORE_NAME,
+                        Duration.ofMillis(1000L),
+                        Duration.ofMillis(1000L),
+                        false),
+                    Serdes.Integer(),
+                    Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(TimestampedWindowedProcessor::new, STORE_NAME);
+
+        final Properties props = props();
+        shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(
+            new KafkaStreams(streamsBuilderForOldStore.build(), props),
+            new KafkaStreams(streamsBuilderForNewStore.build(), props),
+            true);
+    }
+
+    private void shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(final KafkaStreams kafkaStreamsOld,
+                                                                           final KafkaStreams kafkaStreamsNew,
+                                                                           final boolean persistentStore) throws Exception {
+        kafkaStreams = kafkaStreamsOld;
+        kafkaStreams.start();
+
+        processWindowedKeyValueAndVerifyPlainCount(1, singletonList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 1L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(1, singletonList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L)));
+        final long lastUpdateKeyOne = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        processWindowedKeyValueAndVerifyPlainCount(2, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L)));
+        final long lastUpdateKeyTwo = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        processWindowedKeyValueAndVerifyPlainCount(3, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L)));
+        final long lastUpdateKeyThree = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        processWindowedKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), 1L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), 2L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), 3L)));
+        final long lastUpdateKeyFour = persistentStore ? -1L : CLUSTER.time.milliseconds() - 1L;
+
+        kafkaStreams.close();
+        kafkaStreams = null;
+
+
+        kafkaStreams = kafkaStreamsNew;
+        kafkaStreams.start();
+
+        verifyWindowedCountWithTimestamp(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L, lastUpdateKeyOne);
+        verifyWindowedCountWithTimestamp(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L, lastUpdateKeyTwo);
+        verifyWindowedCountWithTimestamp(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L, lastUpdateKeyThree);
+        verifyWindowedCountWithTimestamp(new Windowed<>(4, new TimeWindow(0L, 1000L)), 3L, lastUpdateKeyFour);
+
+        final long currentTime = CLUSTER.time.milliseconds();
+        processKeyValueAndVerifyWindowedCountWithTimestamp(1, currentTime + 42L, asList(
+            KeyValue.pair(
+                new Windowed<>(1, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(
+                new Windowed<>(2, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(1L, lastUpdateKeyTwo)),
+            KeyValue.pair(
+                new Windowed<>(3, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(
+                new Windowed<>(4, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, lastUpdateKeyFour))));
+
+        processKeyValueAndVerifyWindowedCountWithTimestamp(2, currentTime + 45L, asList(
+            KeyValue.pair(
+                new Windowed<>(1, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(
+                new Windowed<>(2, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(
+                new Windowed<>(3, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(
+                new Windowed<>(4, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, lastUpdateKeyFour))));
+
+        // can process "out of order" record for different key
+        processKeyValueAndVerifyWindowedCountWithTimestamp(4, currentTime + 21L, asList(
+            KeyValue.pair(
+                new Windowed<>(1, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(
+                new Windowed<>(2, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(
+                new Windowed<>(3, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(
+                new Windowed<>(4, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(4L, currentTime + 21L))));
+
+        processKeyValueAndVerifyWindowedCountWithTimestamp(4, currentTime + 42L, asList(
+            KeyValue.pair(
+                new Windowed<>(1, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(
+                new Windowed<>(2, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(
+                new Windowed<>(3, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(
+                new Windowed<>(4, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(5L, currentTime + 42L))));
+
+        // out of order (same key) record should not reduce result timestamp
+        processKeyValueAndVerifyWindowedCountWithTimestamp(4, currentTime + 10L, asList(
+            KeyValue.pair(
+                new Windowed<>(1, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(3L, currentTime + 42L)),
+            KeyValue.pair(
+                new Windowed<>(2, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(2L, currentTime + 45L)),
+            KeyValue.pair(
+                new Windowed<>(3, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(1L, lastUpdateKeyThree)),
+            KeyValue.pair(
+                new Windowed<>(4, new TimeWindow(0L, 1000L)),
+                ValueAndTimestamp.make(6L, currentTime + 42L))));
+
+        // test new segment
+        processKeyValueAndVerifyWindowedCountWithTimestamp(10, currentTime + 100001L, singletonList(
+            KeyValue.pair(
+                new Windowed<>(10, new TimeWindow(100000L, 101000L)), ValueAndTimestamp.make(1L, currentTime + 100001L))));
+
+
+        kafkaStreams.close();
+    }
+
+    @Test
+    public void shouldProxyWindowStoreToTimestampedWindowStoreUsingPapi() throws Exception {
+        final StreamsBuilder streamsBuilderForOldStore = new StreamsBuilder();
+
+        streamsBuilderForOldStore.addStateStore(
+            Stores.windowStoreBuilder(
+                Stores.persistentWindowStore(
+                    STORE_NAME,
+                    Duration.ofMillis(1000L),
+                    Duration.ofMillis(1000L),
+                    false),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(WindowedProcessor::new, STORE_NAME);
+
+        final Properties props = props();
+        kafkaStreams = new KafkaStreams(streamsBuilderForOldStore.build(), props);
+        kafkaStreams.start();
+
+        processWindowedKeyValueAndVerifyPlainCount(1, singletonList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 1L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(1, singletonList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(2, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(3, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), 1L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), 2L)));
+
+        processWindowedKeyValueAndVerifyPlainCount(4, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), 3L)));
+
+        kafkaStreams.close();
+        kafkaStreams = null;
+
+
+
+        final StreamsBuilder streamsBuilderForNewStore = new StreamsBuilder();
+
+        streamsBuilderForNewStore.addStateStore(
+            Stores.timestampedWindowStoreBuilder(
+                Stores.persistentWindowStore(
+                    STORE_NAME,
+                    Duration.ofMillis(1000L),
+                    Duration.ofMillis(1000L),
+                    false),
+                Serdes.Integer(),
+                Serdes.Long()))
+            .<Integer, Integer>stream(inputStream)
+            .process(TimestampedWindowedProcessor::new, STORE_NAME);
+
+        kafkaStreams = new KafkaStreams(streamsBuilderForNewStore.build(), props);
+        kafkaStreams.start();
+
+        verifyWindowedCountWithSurrogateTimestamp(new Windowed<>(1, new TimeWindow(0L, 1000L)), 2L);
+        verifyWindowedCountWithSurrogateTimestamp(new Windowed<>(2, new TimeWindow(0L, 1000L)), 1L);
+        verifyWindowedCountWithSurrogateTimestamp(new Windowed<>(3, new TimeWindow(0L, 1000L)), 1L);
+        verifyWindowedCountWithSurrogateTimestamp(new Windowed<>(4, new TimeWindow(0L, 1000L)), 3L);
+
+        processKeyValueAndVerifyWindowedCountWithTimestamp(1, 42L, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L))));
+
+        processKeyValueAndVerifyWindowedCountWithTimestamp(2, 45L, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L))));
+
+        // can process "out of order" record for different key
+        processKeyValueAndVerifyWindowedCountWithTimestamp(4, 21L, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(4L, -1L))));
+
+        processKeyValueAndVerifyWindowedCountWithTimestamp(4, 42L, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(5L, -1L))));
+
+        // out of order (same key) record should not reduce result timestamp
+        processKeyValueAndVerifyWindowedCountWithTimestamp(4, 10L, asList(
+            KeyValue.pair(new Windowed<>(1, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(3L, -1L)),
+            KeyValue.pair(new Windowed<>(2, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(2L, -1L)),
+            KeyValue.pair(new Windowed<>(3, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(1L, -1L)),
+            KeyValue.pair(new Windowed<>(4, new TimeWindow(0L, 1000L)), ValueAndTimestamp.make(6L, -1L))));
+
+        // test new segment
+        processKeyValueAndVerifyWindowedCountWithTimestamp(10, 100001L, singletonList(
+            KeyValue.pair(new Windowed<>(10, new TimeWindow(100000L, 101000L)), ValueAndTimestamp.make(1L, -1L))));
+
+
+        kafkaStreams.close();
+    }
+
+    private <K, V> void processWindowedKeyValueAndVerifyPlainCount(final K key,
+                                                                   final List<KeyValue<Windowed<Integer>, Object>> expectedStoreContent)
+            throws Exception {
+
+        IntegrationTestUtils.produceKeyValuesSynchronously(
+            inputStream,
+            singletonList(KeyValue.pair(key, 0)),
+            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+                IntegerSerializer.class,
+                IntegerSerializer.class),
+            CLUSTER.time);
+
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyWindowStore<K, V> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.windowStore());
+                try (final KeyValueIterator<Windowed<K>, V> all = store.all()) {
+                    final List<KeyValue<Windowed<K>, V>> storeContent = new LinkedList<>();
+                    while (all.hasNext()) {
+                        storeContent.add(all.next());
+                    }
+                    return storeContent.equals(expectedStoreContent);
+                }
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K> void verifyWindowedCountWithSurrogateTimestamp(final Windowed<K> key,
+                                                               final long value) throws Exception {
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyWindowStore<K, ValueAndTimestamp<Long>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedWindowStore());
+                final ValueAndTimestamp<Long> count = store.fetch(key.key(), key.window().start());
+                return count.value() == value && count.timestamp() == -1L;
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K> void verifyWindowedCountWithTimestamp(final Windowed<K> key,
+                                                      final long value,
+                                                      final long timestamp) throws Exception {
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyWindowStore<K, ValueAndTimestamp<Long>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedWindowStore());
+                final ValueAndTimestamp<Long> count = store.fetch(key.key(), key.window().start());
+                return count.value() == value && count.timestamp() == timestamp;
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private <K, V> void processKeyValueAndVerifyWindowedCountWithTimestamp(final K key,
+                                                                           final long timestamp,
+                                                                           final List<KeyValue<Windowed<Integer>, Object>> expectedStoreContent)
+            throws Exception {
+
+        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+            inputStream,
+            singletonList(KeyValue.pair(key, 0)),
+            TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+                IntegerSerializer.class,
+                IntegerSerializer.class),
+            timestamp);
+
+        TestUtils.waitForCondition(() -> {
+            try {
+                final ReadOnlyWindowStore<K, ValueAndTimestamp<V>> store =
+                    kafkaStreams.store(STORE_NAME, QueryableStoreTypes.timestampedWindowStore());
+                try (final KeyValueIterator<Windowed<K>, ValueAndTimestamp<V>> all = store.all()) {
+                    final List<KeyValue<Windowed<K>, ValueAndTimestamp<V>>> storeContent = new LinkedList<>();
+                    while (all.hasNext()) {
+                        storeContent.add(all.next());
+                    }
+                    return storeContent.equals(expectedStoreContent);
+                }
+            } catch (final Exception swallow) {
+                swallow.printStackTrace();
+                System.err.println(swallow.getMessage());
+                return false;
+            }
+        }, "Could not get expected result in time.");
+    }
+
+    private static class KeyValueProcessor implements Processor<Integer, Integer> {
+        private KeyValueStore<Integer, Long> store;
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public void init(final ProcessorContext context) {
+            store = (KeyValueStore<Integer, Long>) context.getStateStore(STORE_NAME);
+        }
+
+        @Override
+        public void process(final Integer key, final Integer value) {
+            final long newCount;
+
+            final Long oldCount = store.get(key);
+            if (oldCount != null) {
+                newCount = oldCount + 1L;
+            } else {
+                newCount = 1L;
+            }
+
+            store.put(key, newCount);
+        }
+
+        @Override
+        public void close() {}
+    }
+
+    private static class TimestampedKeyValueProcessor implements Processor<Integer, Integer> {
+        private ProcessorContext context;
+        private TimestampedKeyValueStore<Integer, Long> store;
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public void init(final ProcessorContext context) {
+            this.context = context;
+            store = (TimestampedKeyValueStore<Integer, Long>) context.getStateStore(STORE_NAME);
+        }
+
+        @Override
+        public void process(final Integer key, final Integer value) {
+            final long newCount;
+
+            final ValueAndTimestamp<Long> oldCountWithTimestamp = store.get(key);
+            final long newTimestamp;
+
+            if (oldCountWithTimestamp == null) {
+                newCount = 1L;
+                newTimestamp = context.timestamp();
+            } else {
+                newCount = oldCountWithTimestamp.value() + 1L;
+                newTimestamp = Math.max(oldCountWithTimestamp.timestamp(), context.timestamp());
+            }
+
+            store.put(key, ValueAndTimestamp.make(newCount, newTimestamp));
+        }
+
+        @Override
+        public void close() {}
+    }
+
+    private static class WindowedProcessor implements Processor<Integer, Integer> {
+        private WindowStore<Integer, Long> store;
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public void init(final ProcessorContext context) {
+            store = (WindowStore<Integer, Long>) context.getStateStore(STORE_NAME);
+        }
+
+        @Override
+        public void process(final Integer key, final Integer value) {
+            final long newCount;
+
+            final Long oldCount = store.fetch(key, key < 10 ? 0L : 100000L);
+            if (oldCount != null) {
+                newCount = oldCount + 1L;
+            } else {
+                newCount = 1L;
+            }
+
+            store.put(key, newCount, key < 10 ? 0L : 100000L);
+        }
+
+        @Override
+        public void close() {}
+    }
+
+    private static class TimestampedWindowedProcessor implements Processor<Integer, Integer> {
+        private ProcessorContext context;
+        private TimestampedWindowStore<Integer, Long> store;
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public void init(final ProcessorContext context) {
+            this.context = context;
+            store = (TimestampedWindowStore<Integer, Long>) context.getStateStore(STORE_NAME);
+        }
+
+        @Override
+        public void process(final Integer key, final Integer value) {
+            final long newCount;
+
+            final ValueAndTimestamp<Long> oldCountWithTimestamp = store.fetch(key, key < 10 ? 0L : 100000L);
+            final long newTimestamp;
+
+            if (oldCountWithTimestamp == null) {
+                newCount = 1L;
+                newTimestamp = context.timestamp();
+            } else {
+                newCount = oldCountWithTimestamp.value() + 1L;
+                newTimestamp = Math.max(oldCountWithTimestamp.timestamp(), context.timestamp());
+            }
+
+            store.put(key, ValueAndTimestamp.make(newCount, newTimestamp), key < 10 ? 0L : 100000L);
+        }
+
+        @Override
+        public void close() {}
+    }
+}
\ No newline at end of file
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
index d577cf7..2394203 100644
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
+++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
@@ -26,7 +26,6 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.LongSerializer;
 import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.Bytes;
@@ -78,6 +77,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -465,23 +465,17 @@ public class TopologyTestDriverTest {
         topology.addSink(
             "sink",
             SINK_TOPIC_1,
-            new Serializer<Object>() {
-                @Override
-                public byte[] serialize(final String topic, final Object data) {
-                    if (data instanceof Long) {
-                        return Serdes.Long().serializer().serialize(topic, (Long) data);
-                    }
-                    return Serdes.Integer().serializer().serialize(topic, (Integer) data);
+            (topic, data) -> {
+                if (data instanceof Long) {
+                    return Serdes.Long().serializer().serialize(topic, (Long) data);
                 }
+                return Serdes.Integer().serializer().serialize(topic, (Integer) data);
             },
-            new Serializer<Object>() {
-                @Override
-                public byte[] serialize(final String topic, final Object data) {
-                    if (data instanceof String) {
-                        return Serdes.String().serializer().serialize(topic, (String) data);
-                    }
-                    return Serdes.Double().serializer().serialize(topic, (Double) data);
+            (topic, data) -> {
+                if (data instanceof String) {
+                    return Serdes.String().serializer().serialize(topic, (String) data);
                 }
+                return Serdes.Double().serializer().serialize(topic, (Double) data);
             },
             processor);
 
@@ -736,6 +730,160 @@ public class TopologyTestDriverTest {
     }
 
     @Test
+    public void shouldReturnCorrectPersistentStoreTypeOnly() {
+        shouldReturnCorrectStoreTypeOnly(true);
+    }
+
+    @Test
+    public void shouldReturnCorrectInMemoryStoreTypeOnly() {
+        shouldReturnCorrectStoreTypeOnly(false);
+    }
+
+    private void shouldReturnCorrectStoreTypeOnly(final boolean persistent) {
+        final String keyValueStoreName = "keyValueStore";
+        final String timestampedKeyValueStoreName = "keyValueTimestampStore";
+        final String windowStoreName = "windowStore";
+        final String timestampedWindowStoreName = "windowTimestampStore";
+        final String sessionStoreName = "sessionStore";
+        final String globalKeyValueStoreName = "globalKeyValueStore";
+        final String globalTimestampedKeyValueStoreName = "globalKeyValueTimestampStore";
+
+        final Topology topology = setupSingleProcessorTopology();
+
+        // add state stores
+        topology.addStateStore(
+            Stores.keyValueStoreBuilder(
+                persistent ?
+                    Stores.persistentKeyValueStore(keyValueStoreName) :
+                    Stores.inMemoryKeyValueStore(keyValueStoreName),
+                Serdes.ByteArray(),
+                Serdes.ByteArray()
+            ),
+            "processor");
+        topology.addStateStore(
+            Stores.timestampedKeyValueStoreBuilder(
+                persistent ?
+                    Stores.persistentTimestampedKeyValueStore(timestampedKeyValueStoreName) :
+                    Stores.inMemoryKeyValueStore(timestampedKeyValueStoreName),
+                Serdes.ByteArray(),
+                Serdes.ByteArray()
+            ),
+            "processor");
+        topology.addStateStore(
+            Stores.windowStoreBuilder(
+                persistent ?
+                    Stores.persistentWindowStore(windowStoreName, Duration.ofMillis(1000L), Duration.ofMillis(100L), false) :
+                    Stores.inMemoryWindowStore(windowStoreName, Duration.ofMillis(1000L), Duration.ofMillis(100L), false),
+                Serdes.ByteArray(),
+                Serdes.ByteArray()
+            ),
+            "processor");
+        topology.addStateStore(
+            Stores.timestampedWindowStoreBuilder(
+                persistent ?
+                    Stores.persistentTimestampedWindowStore(timestampedWindowStoreName, Duration.ofMillis(1000L), Duration.ofMillis(100L), false) :
+                    Stores.inMemoryWindowStore(timestampedWindowStoreName, Duration.ofMillis(1000L), Duration.ofMillis(100L), false),
+                Serdes.ByteArray(),
+                Serdes.ByteArray()
+            ),
+            "processor");
+        topology.addStateStore(
+            persistent ?
+                Stores.sessionStoreBuilder(
+                    Stores.persistentSessionStore(sessionStoreName, Duration.ofMillis(1000L)),
+                    Serdes.ByteArray(),
+                    Serdes.ByteArray()) :
+                Stores.sessionStoreBuilder(
+                    Stores.inMemorySessionStore(sessionStoreName, Duration.ofMillis(1000L)),
+                    Serdes.ByteArray(),
+                    Serdes.ByteArray()),
+            "processor");
+        // add global stores
+        topology.addGlobalStore(
+            persistent ?
+                Stores.keyValueStoreBuilder(
+                    Stores.persistentKeyValueStore(globalKeyValueStoreName),
+                    Serdes.ByteArray(),
+                    Serdes.ByteArray()
+                ).withLoggingDisabled() :
+                Stores.keyValueStoreBuilder(
+                    Stores.inMemoryKeyValueStore(globalKeyValueStoreName),
+                    Serdes.ByteArray(),
+                    Serdes.ByteArray()
+                ).withLoggingDisabled(),
+            "sourceDummy1",
+            Serdes.ByteArray().deserializer(),
+            Serdes.ByteArray().deserializer(),
+            "topicDummy1",
+            "processorDummy1",
+            () -> null);
+        topology.addGlobalStore(
+            persistent ?
+                Stores.timestampedKeyValueStoreBuilder(
+                    Stores.persistentTimestampedKeyValueStore(globalTimestampedKeyValueStoreName),
+                    Serdes.ByteArray(),
+                    Serdes.ByteArray()
+                ).withLoggingDisabled() :
+                Stores.timestampedKeyValueStoreBuilder(
+                    Stores.inMemoryKeyValueStore(globalTimestampedKeyValueStoreName),
+                    Serdes.ByteArray(),
+                    Serdes.ByteArray()
+                ).withLoggingDisabled(),
+            "sourceDummy2",
+            Serdes.ByteArray().deserializer(),
+            Serdes.ByteArray().deserializer(),
+            "topicDummy2",
+            "processorDummy2",
+            () -> null);
+
+        testDriver = new TopologyTestDriver(topology, config);
+
+        // verify state stores
+        assertNotNull(testDriver.getKeyValueStore(keyValueStoreName));
+        assertNull(testDriver.getTimestampedKeyValueStore(keyValueStoreName));
+        assertNull(testDriver.getWindowStore(keyValueStoreName));
+        assertNull(testDriver.getTimestampedWindowStore(keyValueStoreName));
+        assertNull(testDriver.getSessionStore(keyValueStoreName));
+
+        assertNotNull(testDriver.getKeyValueStore(timestampedKeyValueStoreName));
+        assertNotNull(testDriver.getTimestampedKeyValueStore(timestampedKeyValueStoreName));
+        assertNull(testDriver.getWindowStore(timestampedKeyValueStoreName));
+        assertNull(testDriver.getTimestampedWindowStore(timestampedKeyValueStoreName));
+        assertNull(testDriver.getSessionStore(timestampedKeyValueStoreName));
+
+        assertNull(testDriver.getKeyValueStore(windowStoreName));
+        assertNull(testDriver.getTimestampedKeyValueStore(windowStoreName));
+        assertNotNull(testDriver.getWindowStore(windowStoreName));
+        assertNull(testDriver.getTimestampedWindowStore(windowStoreName));
+        assertNull(testDriver.getSessionStore(windowStoreName));
+
+        assertNull(testDriver.getKeyValueStore(timestampedWindowStoreName));
+        assertNull(testDriver.getTimestampedKeyValueStore(timestampedWindowStoreName));
+        assertNotNull(testDriver.getWindowStore(timestampedWindowStoreName));
+        assertNotNull(testDriver.getTimestampedWindowStore(timestampedWindowStoreName));
+        assertNull(testDriver.getSessionStore(timestampedWindowStoreName));
+
+        assertNull(testDriver.getKeyValueStore(sessionStoreName));
+        assertNull(testDriver.getTimestampedKeyValueStore(sessionStoreName));
+        assertNull(testDriver.getWindowStore(sessionStoreName));
+        assertNull(testDriver.getTimestampedWindowStore(sessionStoreName));
+        assertNotNull(testDriver.getSessionStore(sessionStoreName));
+
+        // verify global stores
+        assertNotNull(testDriver.getKeyValueStore(globalKeyValueStoreName));
+        assertNull(testDriver.getTimestampedKeyValueStore(globalKeyValueStoreName));
+        assertNull(testDriver.getWindowStore(globalKeyValueStoreName));
+        assertNull(testDriver.getTimestampedWindowStore(globalKeyValueStoreName));
+        assertNull(testDriver.getSessionStore(globalKeyValueStoreName));
+
+        assertNotNull(testDriver.getKeyValueStore(globalTimestampedKeyValueStoreName));
+        assertNotNull(testDriver.getTimestampedKeyValueStore(globalTimestampedKeyValueStoreName));
+        assertNull(testDriver.getWindowStore(globalTimestampedKeyValueStoreName));
+        assertNull(testDriver.getTimestampedWindowStore(globalTimestampedKeyValueStoreName));
+        assertNull(testDriver.getSessionStore(globalTimestampedKeyValueStoreName));
+    }
+
+    @Test
     public void shouldReturnAllStoresNames() {
         final Topology topology = setupSourceSinkTopology();
         topology.addStateStore(
@@ -793,7 +941,7 @@ public class TopologyTestDriverTest {
         setup();
         testDriver.pipeInput(recordFactory.create("input-topic", "a", 1L, 9999L));
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 21L);
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
     }
 
     @Test
@@ -802,7 +950,7 @@ public class TopologyTestDriverTest {
         testDriver.pipeInput(recordFactory.create("input-topic", "a", 1L, 9999L));
         assertThat(store.get("a"), equalTo(21L));
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 21L);
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
     }
 
     @Test
@@ -811,7 +959,7 @@ public class TopologyTestDriverTest {
         testDriver.pipeInput(recordFactory.create("input-topic", "a", 42L, 9999L));
         assertThat(store.get("a"), equalTo(42L));
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 42L);
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
     }
 
     @Test
@@ -821,7 +969,7 @@ public class TopologyTestDriverTest {
         assertThat(store.get("b"), equalTo(21L));
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 21L);
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "b", 21L);
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
     }
 
     @Test
@@ -831,11 +979,11 @@ public class TopologyTestDriverTest {
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 21L);
 
         testDriver.pipeInput(recordFactory.create("input-topic", "a", 1L, 9999L));
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
 
         testDriver.pipeInput(recordFactory.create("input-topic", "a", 1L, 10000L));
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 21L);
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
     }
 
     @Test
@@ -843,7 +991,7 @@ public class TopologyTestDriverTest {
         setup();
         testDriver.advanceWallClockTime(60000);
         OutputVerifier.compareKeyValue(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer), "a", 21L);
-        Assert.assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
+        assertNull(testDriver.readOutput("result-topic", stringDeserializer, longDeserializer));
     }
 
     private class CustomMaxAggregatorSupplier implements ProcessorSupplier<String, Long> {
@@ -944,14 +1092,14 @@ public class TopologyTestDriverTest {
         config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass().getName());
 
         try (final TopologyTestDriver testDriver = new TopologyTestDriver(topology, config)) {
-            Assert.assertNull(testDriver.getKeyValueStore("storeProcessorStore").get("a"));
+            assertNull(testDriver.getKeyValueStore("storeProcessorStore").get("a"));
             testDriver.pipeInput(recordFactory.create("input-topic", "a", 1L));
             Assert.assertEquals(1L, testDriver.getKeyValueStore("storeProcessorStore").get("a"));
         }
 
 
         try (final TopologyTestDriver testDriver = new TopologyTestDriver(topology, config)) {
-            Assert.assertNull(
+            assertNull(
                 "Closing the prior test driver should have cleaned up this store and value.",
                 testDriver.getKeyValueStore("storeProcessorStore").get("a")
             );


Mime
View raw message