content: /*
   * 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.api.Processor;
  import org.apache.kafka.streams.processor.api.ProcessorContext;
  import org.apache.kafka.streams.processor.api.Record;
  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.TestUtils;
  
  import org.junit.jupiter.api.AfterAll;
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeAll;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Tag;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.TestInfo;
  
  import java.io.IOException;
  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;
  import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
  
  @Tag("integration")
  public class StoreUpgradeIntegrationTest {
      private static final String STORE_NAME = "store";
      private String inputStream;
  
      private KafkaStreams kafkaStreams;
  
      public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
  
      @BeforeAll
      public static void startCluster() throws IOException {
          CLUSTER.start();
      }
  
      @AfterAll
      public static void closeCluster() {
          CLUSTER.stop();
      }
  
      public String safeTestName;
  
      @BeforeEach
      public void createTopics(final TestInfo testInfo) throws Exception {
          safeTestName = safeUniqueTestName(testInfo);
          inputStream = "input-stream-" + safeTestName;
          CLUSTER.createTopic(inputStream);
      }
  
      private Properties props() {
          final Properties streamsConfiguration = new Properties();
          streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName);
          streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
          streamsConfiguration.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_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, 1000L);
          streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          return streamsConfiguration;
      }
  
      @AfterEach
      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 = IntegrationTestUtils.getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.keyValueStore());
  
                      if (store == null) {
                          return false;
                      }
  
                      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;
                  }
              },
              60_000L,
              "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedKeyValueStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              5_000L,
              () -> "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedKeyValueStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedKeyValueStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedKeyValueStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "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(
              streamsBuilderForOldStore,
              streamsBuilderForNewStore,
              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);
  
          shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(
              streamsBuilderForOldStore,
              streamsBuilderForNewStore,
              true);
      }
  
      private void shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(final StreamsBuilder streamsBuilderForOldStore,
                                                                             final StreamsBuilder streamsBuilderForNewStore,
                                                                             final boolean persistentStore) throws Exception {
          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)));
          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 = new KafkaStreams(streamsBuilderForNewStore.build(), props);
          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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "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 = IntegrationTestUtils
                          .getStore(STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
  
                      if (store == null)
                          return false;
  
                      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;
                  }
              },
              60_000L,
              "Could not get expected result in time.");
      }
  
      private static class KeyValueProcessor implements Processor<Integer, Integer, Void, Void> {
          private KeyValueStore<Integer, Long> store;
  
          @Override
          public void init(final ProcessorContext<Void, Void> context) {
              store = context.getStateStore(STORE_NAME);
          }
  
          @Override
          public void process(final Record<Integer, Integer> record) {
              final long newCount;
  
              final Long oldCount = store.get(record.key());
              if (oldCount != null) {
                  newCount = oldCount + 1L;
              } else {
                  newCount = 1L;
              }
  
              store.put(record.key(), newCount);
          }
  
      }
  
      private static class TimestampedKeyValueProcessor implements Processor<Integer, Integer, Void, Void> {
          private TimestampedKeyValueStore<Integer, Long> store;
  
          @Override
          public void init(final ProcessorContext<Void, Void> context) {
              store = context.getStateStore(STORE_NAME);
          }
  
          @Override
          public void process(final Record<Integer, Integer> record) {
              final long newCount;
  
              final ValueAndTimestamp<Long> oldCountWithTimestamp = store.get(record.key());
              final long newTimestamp;
  
              if (oldCountWithTimestamp == null) {
                  newCount = 1L;
                  newTimestamp = record.timestamp();
              } else {
                  newCount = oldCountWithTimestamp.value() + 1L;
                  newTimestamp = Math.max(oldCountWithTimestamp.timestamp(), record.timestamp());
              }
  
              store.put(record.key(), ValueAndTimestamp.make(newCount, newTimestamp));
          }
  
      }
  
      private static class WindowedProcessor implements Processor<Integer, Integer, Void, Void> {
          private WindowStore<Integer, Long> store;
  
          @Override
          public void init(final ProcessorContext<Void, Void> context) {
              store = context.getStateStore(STORE_NAME);
          }
  
          @Override
          public void process(final Record<Integer, Integer> record) {
              final long newCount;
  
              final Long oldCount = store.fetch(record.key(), record.key() < 10 ? 0L : 100000L);
              if (oldCount != null) {
                  newCount = oldCount + 1L;
              } else {
                  newCount = 1L;
              }
  
              store.put(record.key(), newCount, record.key() < 10 ? 0L : 100000L);
          }
  
      }
  
      private static class TimestampedWindowedProcessor implements Processor<Integer, Integer, Void, Void> {
          private TimestampedWindowStore<Integer, Long> store;
  
          @Override
          public void init(final ProcessorContext<Void, Void> context) {
              store = context.getStateStore(STORE_NAME);
          }
  
          @Override
          public void process(final Record<Integer, Integer> record) {
              final long newCount;
  
              final ValueAndTimestamp<Long> oldCountWithTimestamp = store.fetch(record.key(), record.key() < 10 ? 0L : 100000L);
              final long newTimestamp;
  
              if (oldCountWithTimestamp == null) {
                  newCount = 1L;
                  newTimestamp = record.timestamp();
              } else {
                  newCount = oldCountWithTimestamp.value() + 1L;
                  newTimestamp = Math.max(oldCountWithTimestamp.timestamp(), record.timestamp());
              }
  
              store.put(record.key(), ValueAndTimestamp.make(newCount, newTimestamp), record.key() < 10 ? 0L : 100000L);
          }
  
      }
  }