/*
 * 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;
        }
    }

}
