/*
 * 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.processor.internals;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler.DeserializationHandlerResponse;
import org.apache.kafka.streams.errors.ErrorHandlerContext;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.test.InternalMockProcessorContext;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;

import java.util.Map;
import java.util.Optional;

import static org.apache.kafka.streams.StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;

public class RecordDeserializerTest {
    private final String sourceNodeName = "source-node";
    private final TaskId taskId = new TaskId(0, 0);
    private final RecordHeaders headers = new RecordHeaders(new Header[]{new RecordHeader("key", "value".getBytes())});
    private final ConsumerRecord<byte[], byte[]> rawRecord = new ConsumerRecord<>("topic",
        1,
        1,
        10,
        TimestampType.LOG_APPEND_TIME,
        3,
        5,
        new byte[0],
        new byte[0],
        headers,
        Optional.of(5));

    private final InternalProcessorContext<Void, Void> context = new InternalMockProcessorContext<>();

    @Test
    public void shouldReturnConsumerRecordWithDeserializedValueWhenNoExceptions() {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            false,
                            false,
                            "key",
                            "value"
                    ),
                    null,
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );
            final ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(null, rawRecord);
            assertEquals(rawRecord.topic(), record.topic());
            assertEquals(rawRecord.partition(), record.partition());
            assertEquals(rawRecord.offset(), record.offset());
            assertEquals("key", record.key());
            assertEquals("value", record.value());
            assertEquals(rawRecord.timestamp(), record.timestamp());
            assertEquals(TimestampType.CREATE_TIME, record.timestampType());
            assertEquals(rawRecord.headers(), record.headers());
            assertEquals(rawRecord.leaderEpoch(), record.leaderEpoch());
        }
    }

    @ParameterizedTest
    @CsvSource({
        "true, true",
        "true, false",
        "false, true",
    })
    public void shouldThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithFail(final boolean keyThrowsException,
                                                                                                      final boolean valueThrowsException) {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            keyThrowsException,
                            valueThrowsException,
                            "key",
                            "value"
                    ),
                    new DeserializationExceptionHandlerMock(
                            Optional.of(DeserializationHandlerResponse.FAIL),
                            rawRecord,
                            sourceNodeName,
                            taskId
                    ),
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );

            final StreamsException e = assertThrows(StreamsException.class, () -> recordDeserializer.deserialize(context, rawRecord));
            assertEquals(
                    e.getMessage(),
                    "Deserialization exception handler is set "
                            + "to fail upon a deserialization error. "
                            + "If you would rather have the streaming pipeline "
                            + "continue after a deserialization error, please set the "
                            + DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG + " appropriately."
            );
        }
    }

    @ParameterizedTest
    @CsvSource({
        "true, true",
        "true, false",
        "false, true"
    })
    public void shouldNotThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithContinue(final boolean keyThrowsException,
                                                                                                             final boolean valueThrowsException) {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            keyThrowsException,
                            valueThrowsException,
                            "key",
                            "value"
                    ),
                    new DeserializationExceptionHandlerMock(
                            Optional.of(DeserializationHandlerResponse.CONTINUE),
                            rawRecord,
                            sourceNodeName,
                            taskId
                    ),
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );

            final ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(context, rawRecord);
            assertNull(record);
        }
    }

    @Test
    public void shouldFailWhenDeserializationFailsAndExceptionHandlerReturnsNull() {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            true,
                            false,
                            "key",
                            "value"
                    ),
                    new DeserializationExceptionHandlerMock(
                            Optional.empty(),
                            rawRecord,
                            sourceNodeName,
                            taskId
                    ),
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );

            final StreamsException exception = assertThrows(
                    StreamsException.class,
                    () -> recordDeserializer.deserialize(context, rawRecord)
            );
            assertEquals("Fatal user code error in deserialization error callback", exception.getMessage());
            assertInstanceOf(NullPointerException.class, exception.getCause());
            assertEquals("Invalid DeserializationExceptionHandler response.", exception.getCause().getMessage());
        }
    }

    @Test
    public void shouldFailWhenDeserializationFailsAndExceptionHandlerThrows() {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            true,
                            false,
                            "key",
                            "value"
                    ),
                    new DeserializationExceptionHandlerMock(
                            null, // indicate to throw an exception
                            rawRecord,
                            sourceNodeName,
                            taskId
                    ),
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );

            final StreamsException exception = assertThrows(
                    StreamsException.class,
                    () -> recordDeserializer.deserialize(context, rawRecord)
            );
            assertEquals("Fatal user code error in deserialization error callback", exception.getMessage());
            assertEquals("CRASH", exception.getCause().getMessage());
        }
    }

    static class TheSourceNode extends SourceNode<Object, Object> {
        private final boolean keyThrowsException;
        private final boolean valueThrowsException;
        private final Object key;
        private final Object value;

        TheSourceNode(final String name,
                      final boolean keyThrowsException,
                      final boolean valueThrowsException,
                      final Object key,
                      final Object value) {
            super(name, null, null);
            this.keyThrowsException = keyThrowsException;
            this.valueThrowsException = valueThrowsException;
            this.key = key;
            this.value = value;
        }

        @Override
        public Object deserializeKey(final String topic, final Headers headers, final byte[] data) {
            if (keyThrowsException) {
                throw new RuntimeException("KABOOM!");
            }
            return key;
        }

        @Override
        public Object deserializeValue(final String topic, final Headers headers, final byte[] data) {
            if (valueThrowsException) {
                throw new RuntimeException("KABOOM!");
            }
            return value;
        }
    }

    public static class DeserializationExceptionHandlerMock implements DeserializationExceptionHandler {
        private final Optional<DeserializationHandlerResponse> response;
        private final ConsumerRecord<byte[], byte[]> expectedRecord;
        private final String expectedProcessorNodeId;
        private final TaskId expectedTaskId;

        public DeserializationExceptionHandlerMock(final Optional<DeserializationHandlerResponse> response,
                                                   final ConsumerRecord<byte[], byte[]> record,
                                                   final String processorNodeId,
                                                   final TaskId taskId) {
            this.response = response;
            this.expectedRecord = record;
            this.expectedProcessorNodeId = processorNodeId;
            this.expectedTaskId = taskId;
        }

        @Override
        public DeserializationHandlerResponse handle(final ErrorHandlerContext context,
                                                     final ConsumerRecord<byte[], byte[]> record,
                                                     final Exception exception) {
            assertEquals(expectedRecord.topic(), context.topic());
            assertEquals(expectedRecord.partition(), context.partition());
            assertEquals(expectedRecord.offset(), context.offset());
            assertEquals(expectedProcessorNodeId, context.processorNodeId());
            assertEquals(expectedTaskId, context.taskId());
            assertEquals(expectedRecord.timestamp(), context.timestamp());
            assertEquals(expectedRecord, record);
            assertInstanceOf(RuntimeException.class, exception);
            assertEquals("KABOOM!", exception.getMessage());
            if (response == null) {
                throw new RuntimeException("CRASH");
            }
            return response.orElse(null);
        }

        @Override
        public void configure(final Map<String, ?> configs) {
            // do nothing
        }
    }
}
