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.clients.consumer.KafkaConsumer;
  import org.apache.kafka.clients.consumer.OffsetAndMetadata;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.serialization.Serde;
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.apache.kafka.server.util.MockTime;
  import org.apache.kafka.streams.AutoOffsetReset;
  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.errors.StreamsUncaughtExceptionHandler;
  import org.apache.kafka.streams.errors.TopologyException;
  import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
  import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
  import org.apache.kafka.streams.kstream.Consumed;
  import org.apache.kafka.streams.kstream.KStream;
  import org.apache.kafka.streams.kstream.Produced;
  import org.apache.kafka.test.StreamsTestUtils;
  import org.apache.kafka.test.TestUtils;
  
  import org.junit.jupiter.api.AfterAll;
  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.Timeout;
  
  import java.io.IOException;
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.regex.Pattern;
  
  import static org.apache.kafka.common.utils.Utils.mkProperties;
  import static org.apache.kafka.test.TestUtils.waitForCondition;
  import static org.hamcrest.CoreMatchers.equalTo;
  import static org.hamcrest.CoreMatchers.is;
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.junit.jupiter.api.Assertions.fail;
  
  @Timeout(600)
  @Tag("integration")
  public class FineGrainedAutoResetIntegrationTest {
      private static final int NUM_BROKERS = 1;
      private static final String DEFAULT_OUTPUT_TOPIC = "outputTopic";
      private static final String OUTPUT_TOPIC_0 = "outputTopic_0";
      private static final String OUTPUT_TOPIC_1 = "outputTopic_1";
      private static final String OUTPUT_TOPIC_2 = "outputTopic_2";
      private static final String OUTPUT_TOPIC_3 = "outputTopic_3";
      private static final String OUTPUT_TOPIC_4 = "outputTopic_4";
      private static final String OUTPUT_TOPIC_5 = "outputTopic_5";
  
      public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
              NUM_BROKERS,
              mkProperties(
                      Collections.singletonMap("log.message.timestamp.after.max.ms", String.valueOf(Long.MAX_VALUE))));
  
      @BeforeAll
      public static void startCluster() throws IOException, InterruptedException {
          CLUSTER.start();
          CLUSTER.createTopics(
                  TOPIC_1_0,
                  TOPIC_2_0,
                  TOPIC_A_0,
                  TOPIC_C_0,
                  TOPIC_Y_0,
                  TOPIC_Z_0,
                  TOPIC_1_1,
                  TOPIC_2_1,
                  TOPIC_A_1,
                  TOPIC_C_1,
                  TOPIC_Y_1,
                  TOPIC_Z_1,
                  TOPIC_1_2,
                  TOPIC_2_2,
                  TOPIC_A_2,
                  TOPIC_C_2,
                  TOPIC_Y_2,
                  TOPIC_Z_2,
                  TOPIC_DURATION_1,
                  TOPIC_DURATION_1,
                  TOPIC_DURATION_3,
                  NOOP,
                  DEFAULT_OUTPUT_TOPIC,
                  OUTPUT_TOPIC_0,
                  OUTPUT_TOPIC_1,
                  OUTPUT_TOPIC_2);
      }
  
      @AfterAll
      public static void closeCluster() {
          CLUSTER.stop();
      }
  
      private final MockTime mockTime = CLUSTER.time;
  
      private static final String TOPIC_1_0 = "topic-1_0";
      private static final String TOPIC_2_0 = "topic-2_0";
      private static final String TOPIC_A_0 = "topic-A_0";
      private static final String TOPIC_C_0 = "topic-C_0";
      private static final String TOPIC_Y_0 = "topic-Y_0";
      private static final String TOPIC_Z_0 = "topic-Z_0";
      private static final String TOPIC_1_1 = "topic-1_1";
      private static final String TOPIC_2_1 = "topic-2_1";
      private static final String TOPIC_A_1 = "topic-A_1";
      private static final String TOPIC_C_1 = "topic-C_1";
      private static final String TOPIC_Y_1 = "topic-Y_1";
      private static final String TOPIC_Z_1 = "topic-Z_1";
      private static final String TOPIC_1_2 = "topic-1_2";
      private static final String TOPIC_2_2 = "topic-2_2";
      private static final String TOPIC_A_2 = "topic-A_2";
      private static final String TOPIC_C_2 = "topic-C_2";
      private static final String TOPIC_Y_2 = "topic-Y_2";
      private static final String TOPIC_Z_2 = "topic-Z_2";
      private static final String TOPIC_DURATION_1 = "durationTopic-1";
      private static final String TOPIC_DURATION_2 = "durationTopic-2";
      private static final String TOPIC_DURATION_3 = "durationTopic-3";
      private static final String NOOP = "noop";
      private final Serde<String> stringSerde = Serdes.String();
  
      private static final String STRING_SERDE_CLASSNAME = Serdes.StringSerde.class.getName();
      private Properties streamsConfiguration;
  
      private final String topic1TestMessage = "topic-1 test";
      private final String topic2TestMessage = "topic-2 test";
      private final String topicATestMessage = "topic-A test";
      private final String topicCTestMessage = "topic-C test";
      private final String topicYTestMessage = "topic-Y test";
      private final String topicZTestMessage = "topic-Z test";
  
      @BeforeEach
      public void setUp() throws Exception {
  
          final Properties props = new Properties();
          props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
          props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
          props.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000");
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  
          streamsConfiguration = StreamsTestUtils.getStreamsConfig(
                  "testAutoOffsetId",
                  CLUSTER.bootstrapServers(),
                  STRING_SERDE_CLASSNAME,
                  STRING_SERDE_CLASSNAME,
                  props);
  
          // Remove any state from previous test runs
          IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
      }
  
      @Test
      public void shouldOnlyReadRecordsWhereEarliestSpecifiedWithNoCommittedOffsetsWithGlobalAutoOffsetResetLatest() throws Exception {
          streamsConfiguration.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "latest");
  
          final List<String> expectedReceivedValues = Arrays.asList(topic1TestMessage, topic2TestMessage);
          shouldOnlyReadForEarliest("_0", TOPIC_1_0, TOPIC_2_0, TOPIC_A_0, TOPIC_C_0, TOPIC_Y_0, TOPIC_Z_0, OUTPUT_TOPIC_0, expectedReceivedValues);
      }
  
      @Test
      public void shouldOnlyReadRecordsWhereEarliestSpecifiedWithNoCommittedOffsetsWithDefaultGlobalAutoOffsetResetEarliest() throws Exception {
          final List<String> expectedReceivedValues = Arrays.asList(topic1TestMessage, topic2TestMessage, topicYTestMessage, topicZTestMessage);
          shouldOnlyReadForEarliest("_1", TOPIC_1_1, TOPIC_2_1, TOPIC_A_1, TOPIC_C_1, TOPIC_Y_1, TOPIC_Z_1, OUTPUT_TOPIC_1, expectedReceivedValues);
      }
  
      @Test
      public void shouldOnlyReadRecordsWhereEarliestSpecifiedWithInvalidCommittedOffsets() throws Exception {
          commitInvalidOffsets();
  
          final List<String> expectedReceivedValues = Arrays.asList(topic1TestMessage, topic2TestMessage, topicYTestMessage, topicZTestMessage);
          shouldOnlyReadForEarliest("_2", TOPIC_1_2, TOPIC_2_2, TOPIC_A_2, TOPIC_C_2, TOPIC_Y_2, TOPIC_Z_2, OUTPUT_TOPIC_2, expectedReceivedValues);
      }
  
      private void shouldOnlyReadForEarliest(
          final String topicSuffix,
          final String topic1,
          final String topic2,
          final String topicA,
          final String topicC,
          final String topicY,
          final String topicZ,
          final String outputTopic,
          final List<String> expectedReceivedValues) throws Exception {
  
          final StreamsBuilder builder = new StreamsBuilder();
  
  
          final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("topic-\\d" + topicSuffix), Consumed.with(AutoOffsetReset.earliest()));
          final KStream<String, String> pattern2Stream = builder.stream(Pattern.compile("topic-[A-D]" + topicSuffix), Consumed.with(AutoOffsetReset.latest()));
          final KStream<String, String> namedTopicsStream = builder.stream(Arrays.asList(topicY, topicZ));
  
          pattern1Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));
          pattern2Stream.to(outputTopic, Produced.with(stringSerde, stringSerde));
          namedTopicsStream.to(outputTopic, Produced.with(stringSerde, stringSerde));
  
          final Properties producerConfig = TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class);
  
          IntegrationTestUtils.produceValuesSynchronously(topic1, Collections.singletonList(topic1TestMessage), producerConfig, mockTime);
          IntegrationTestUtils.produceValuesSynchronously(topic2, Collections.singletonList(topic2TestMessage), producerConfig, mockTime);
          IntegrationTestUtils.produceValuesSynchronously(topicA, Collections.singletonList(topicATestMessage), producerConfig, mockTime);
          IntegrationTestUtils.produceValuesSynchronously(topicC, Collections.singletonList(topicCTestMessage), producerConfig, mockTime);
          IntegrationTestUtils.produceValuesSynchronously(topicY, Collections.singletonList(topicYTestMessage), producerConfig, mockTime);
          IntegrationTestUtils.produceValuesSynchronously(topicZ, Collections.singletonList(topicZTestMessage), producerConfig, mockTime);
  
          final Properties consumerConfig = TestUtils.consumerConfig(CLUSTER.bootstrapServers(), StringDeserializer.class, StringDeserializer.class);
  
          final List<String> actualValues = new ArrayList<>(expectedReceivedValues.size());
  
          try (final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration)) {
              streams.start();
  
              final List<KeyValue<String, String>> receivedKeyValues = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, outputTopic, expectedReceivedValues.size());
  
              for (final KeyValue<String, String> receivedKeyValue : receivedKeyValues) {
                  actualValues.add(receivedKeyValue.value);
              }
          }
  
          Collections.sort(actualValues);
          Collections.sort(expectedReceivedValues);
          assertThat(actualValues, equalTo(expectedReceivedValues));
      }
  
      private void commitInvalidOffsets() {
          final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(TestUtils.consumerConfig(
              CLUSTER.bootstrapServers(),
              "commit_invalid_offset_app", // Having a separate application id to avoid waiting for last test poll interval timeout.
              StringDeserializer.class,
              StringDeserializer.class));
  
          final Map<TopicPartition, OffsetAndMetadata> invalidOffsets = new HashMap<>();
          invalidOffsets.put(new TopicPartition(TOPIC_1_2, 0), new OffsetAndMetadata(5, null));
          invalidOffsets.put(new TopicPartition(TOPIC_2_2, 0), new OffsetAndMetadata(5, null));
          invalidOffsets.put(new TopicPartition(TOPIC_A_2, 0), new OffsetAndMetadata(5, null));
          invalidOffsets.put(new TopicPartition(TOPIC_C_2, 0), new OffsetAndMetadata(5, null));
          invalidOffsets.put(new TopicPartition(TOPIC_Y_2, 0), new OffsetAndMetadata(5, null));
          invalidOffsets.put(new TopicPartition(TOPIC_Z_2, 0), new OffsetAndMetadata(5, null));
  
          consumer.commitSync(invalidOffsets);
  
          consumer.close();
      }
  
      @Test
      public void shouldFailForResetNone() throws Exception {
          final Properties props = new Properties();
          props.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000");
  
          final Properties localConfig = StreamsTestUtils.getStreamsConfig(
              "testConfigAutoOffsetWithNone",
              CLUSTER.bootstrapServers(),
              STRING_SERDE_CLASSNAME,
              STRING_SERDE_CLASSNAME,
              props);
  
          final StreamsBuilder builder = new StreamsBuilder();
          final KStream<String, String> exceptionStream = builder.stream(NOOP, Consumed.with(AutoOffsetReset.none()));
  
          exceptionStream.to(DEFAULT_OUTPUT_TOPIC, Produced.with(stringSerde, stringSerde));
  
          try (final KafkaStreams streams = new KafkaStreams(builder.build(), localConfig)) {
              final TestingUncaughtExceptionHandler uncaughtExceptionHandler = new TestingUncaughtExceptionHandler();
              streams.setUncaughtExceptionHandler(uncaughtExceptionHandler);
  
              streams.start();
  
              waitForCondition(
                  () -> uncaughtExceptionHandler.correctExceptionThrown,
                  "The expected NoOffsetForPartitionException was never thrown"
              );
          }
      }
  
      @Test
      public void shouldResetByDuration() throws Exception {
          final StreamsBuilder builder = new StreamsBuilder();
  
          builder.<String, String>stream(TOPIC_DURATION_1, Consumed.with(AutoOffsetReset.byDuration(Duration.ofHours(6L).plus(Duration.ofMinutes(40L)))))
              .to(OUTPUT_TOPIC_3, Produced.with(stringSerde, stringSerde));
          builder.<String, String>stream(TOPIC_DURATION_2, Consumed.with(AutoOffsetReset.byDuration(Duration.ofMillis(mockTime.milliseconds()).minus(Duration.ofDays(1L)))))
              .to(OUTPUT_TOPIC_4, Produced.with(stringSerde, stringSerde));
          builder.<String, String>stream(TOPIC_DURATION_3, Consumed.with(AutoOffsetReset.byDuration(Duration.ZERO)))
              .to(OUTPUT_TOPIC_5, Produced.with(stringSerde, stringSerde));
  
          final Properties producerConfig = TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class);
  
          for (int i = 0; i < 10; ++i) {
              mockTime.sleep(Duration.ofHours(1L).toMillis());
              IntegrationTestUtils.produceValuesSynchronously(TOPIC_DURATION_1, Collections.singletonList("" + i), producerConfig, mockTime);
              IntegrationTestUtils.produceValuesSynchronously(TOPIC_DURATION_2, Collections.singletonList("" + i), producerConfig, mockTime);
              IntegrationTestUtils.produceValuesSynchronously(TOPIC_DURATION_3, Collections.singletonList("" + i), producerConfig, mockTime);
          }
  
          final Properties consumerConfig = TestUtils.consumerConfig(CLUSTER.bootstrapServers(), StringDeserializer.class, StringDeserializer.class);
  
          final List<String> expectedValues = Arrays.asList("3", "4", "5", "6", "7", "8", "9");
          final List<String> allExpectedValues = Arrays.asList("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
          final List<String> singleFinalExpectedValues = List.of("10");
          final List<String> actualValuesOne = new ArrayList<>(expectedValues.size());
          final List<String> actualValuesTwo = new ArrayList<>(allExpectedValues.size());
          final List<String> actualValuesThree = new ArrayList<>(singleFinalExpectedValues.size());
  
          final MockTime streamsMockTime = new MockTime(mockTime.milliseconds() + Duration.ofMinutes(20).toMillis(), 0);
          try (final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration, streamsMockTime)) {
              streams.start();
  
              final List<KeyValue<String, String>> receivedKeyValuesOne =
                  IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_TOPIC_3, expectedValues.size());
              for (final KeyValue<String, String> receivedKeyValue : receivedKeyValuesOne) {
                  actualValuesOne.add(receivedKeyValue.value);
              }
  
              final List<KeyValue<String, String>> receivedKeyValuesTwo =
                  IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_TOPIC_4, allExpectedValues.size());
              for (final KeyValue<String, String> receivedKeyValue : receivedKeyValuesTwo) {
                  actualValuesTwo.add(receivedKeyValue.value);
              }
  
              IntegrationTestUtils.produceValuesSynchronously(TOPIC_DURATION_3, Collections.singletonList("10"), producerConfig, mockTime);
              final List<KeyValue<String, String>> receivedKeyValuesThree =
                  IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_TOPIC_5, singleFinalExpectedValues.size());
              for (final KeyValue<String, String> receivedKeyValue : receivedKeyValuesThree) {
                  actualValuesThree.add(receivedKeyValue.value);
              }
          }
  
          Collections.sort(actualValuesOne);
          Collections.sort(expectedValues);
          assertThat(actualValuesOne, equalTo(expectedValues));
  
          Collections.sort(actualValuesTwo);
          Collections.sort(allExpectedValues);
          assertThat(actualValuesTwo, equalTo(allExpectedValues));
  
          assertThat(actualValuesThree, equalTo(singleFinalExpectedValues));
      }
  
      @Test
      public void shouldThrowExceptionOverlappingPattern() {
          final StreamsBuilder builder = new StreamsBuilder();
          //NOTE this would realistically get caught when building topology, the test is for completeness
          builder.stream(Pattern.compile("topic-[A-D]_1"), Consumed.with(AutoOffsetReset.earliest()));
  
          try {
              builder.stream(Pattern.compile("topic-[A-D]_1"), Consumed.with(AutoOffsetReset.latest()));
              builder.build();
              fail("Should have thrown TopologyException");
          } catch (final TopologyException expected) {
              // do nothing
          }
      }
  
      @Test
      public void shouldThrowExceptionOverlappingTopic() {
          final StreamsBuilder builder = new StreamsBuilder();
          //NOTE this would realistically get caught when building topology, the test is for completeness
          builder.stream(Pattern.compile("topic-[A-D]_1"), Consumed.with(AutoOffsetReset.earliest()));
          try {
              builder.stream(Arrays.asList(TOPIC_A_1, TOPIC_Z_1), Consumed.with(AutoOffsetReset.latest()));
              builder.build();
              fail("Should have thrown TopologyException");
          } catch (final TopologyException expected) {
              // do nothing
          }
      }
  
      @Test
      public void shouldThrowStreamsExceptionNoResetSpecified() throws InterruptedException {
          final Properties props = new Properties();
          props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
          props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
          props.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000");
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
  
          final Properties localConfig = StreamsTestUtils.getStreamsConfig(
                  "testAutoOffsetWithNone",
                  CLUSTER.bootstrapServers(),
                  STRING_SERDE_CLASSNAME,
                  STRING_SERDE_CLASSNAME,
                  props);
  
          final StreamsBuilder builder = new StreamsBuilder();
          final KStream<String, String> exceptionStream = builder.stream(NOOP);
  
          exceptionStream.to(DEFAULT_OUTPUT_TOPIC, Produced.with(stringSerde, stringSerde));
  
          try (final KafkaStreams streams = new KafkaStreams(builder.build(), localConfig)) {
              final TestingUncaughtExceptionHandler uncaughtExceptionHandler = new TestingUncaughtExceptionHandler();
              streams.setUncaughtExceptionHandler(uncaughtExceptionHandler);
  
              streams.start();
  
              waitForCondition(
                  () -> uncaughtExceptionHandler.correctExceptionThrown,
                  "The expected NoOffsetForPartitionException was never thrown"
              );
          }
      }
  
  
      private static final class TestingUncaughtExceptionHandler implements StreamsUncaughtExceptionHandler {
          boolean correctExceptionThrown = false;
          @Override
          public StreamThreadExceptionResponse handle(final Throwable throwable) {
              assertThat(throwable.getClass().getSimpleName(), is("StreamsException"));
              assertThat(throwable.getCause().getClass().getSimpleName(), is("NoOffsetForPartitionException"));
              correctExceptionThrown = true;
              return StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
          }
      }
  
  }
  