/*
 * 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.camel.kafkaconnector;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import org.apache.camel.LoggingLevel;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.kafkaconnector.utils.StringJoinerAggregator;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.header.Header;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.jupiter.api.Test;

import static org.apache.camel.util.CollectionHelper.mapOf;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class CamelSourceTaskTest {
    private static final String DIRECT_URI = "direct:start";
    private static final String TOPIC_NAME = "my-topic";

    private void sendBatchOfRecords(CamelSourceTask sourceTask, long size) {
        final ProducerTemplate template = sourceTask.getCms().getProducerTemplate();
        for (int i = 0; i < size; i++) {
            template.sendBody(DIRECT_URI, "test" + i);
        }
    }

    @Test
    public void testSourcePolling() {
        final long size = 2;
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sendBatchOfRecords(sourceTask, size);
        List<SourceRecord> poll = sourceTask.poll();

        assertEquals(size, poll.size());
        assertEquals(TOPIC_NAME, poll.get(0).topic());
        assertNotNull(poll.get(0).timestamp());
        assertEquals(LoggingLevel.OFF.toString(), sourceTask.getCamelSourceConnectorConfig(props)
            .getString(CamelSourceConnectorConfig.CAMEL_SOURCE_CONTENT_LOG_LEVEL_CONF));

        sourceTask.stop();
    }

    @Test
    public void testSourcePollingMaxNotCommittedRecords() {
        final long size = 4;
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MAX_NOT_COMMITTED_RECORDS_CONF, String.valueOf(size));

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sendBatchOfRecords(sourceTask, size + 1);
        List<SourceRecord> poll = sourceTask.poll();

        assertEquals(4, poll.size());
        sourceTask.stop();
    }

    @Test
    public void testSourcePollingMaxBatchPollSize() {
        final long size = 2;
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MAX_BATCH_POLL_SIZE_CONF, String.valueOf(size));

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sendBatchOfRecords(sourceTask, size + 1);
        List<SourceRecord> poll = sourceTask.poll();
        int pollSize = poll.size();

        assertTrue(pollSize >= 0 && pollSize <= size, "Batch size: " + pollSize + ", expected between 0 and " + size);
        sourceTask.stop();
    }

    @Test
    public void testSourcePollingTimeout() {
        final long size = 999;
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MAX_POLL_DURATION_CONF, "2");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sendBatchOfRecords(sourceTask, size);
        List<SourceRecord>  poll = sourceTask.poll();
        int pollSize = poll.size();

        assertTrue(pollSize < size, "Batch size: " + pollSize + ", expected strictly less than " + size);
        sourceTask.stop();
    }

    @Test
    public void testSourcePollingWithKey() {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MESSAGE_HEADER_KEY_CONF, "CamelSpecialTestKey");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);
        final ProducerTemplate template = sourceTask.getCms().getProducerTemplate();

        // key in the message with body
        template.sendBodyAndHeader(DIRECT_URI, "test", "CamelSpecialTestKey", 1234);

        List<SourceRecord> poll1 = sourceTask.poll();
        assertEquals(1, poll1.size());
        assertEquals(1234, poll1.get(0).key());
        assertEquals(Schema.Type.INT32, poll1.get(0).keySchema().type());

        // no key under the header
        template.sendBodyAndHeader(DIRECT_URI, "test", "WrongHeader", 1234);

        List<SourceRecord> poll2 = sourceTask.poll();
        assertEquals(1, poll2.size());
        assertNull(poll2.get(0).key());
        assertNull(poll2.get(0).keySchema());

        // header with null value
        template.sendBodyAndHeader(DIRECT_URI, "test", "CamelSpecialTestKey", null);

        List<SourceRecord> poll3 = sourceTask.poll();
        assertEquals(1, poll3.size());
        assertNull(poll3.get(0).key());
        assertNull(poll3.get(0).keySchema());

        sourceTask.stop();
    }

    @Test
    public void testSourcePollingWithBody() {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);
        final ProducerTemplate template = sourceTask.getCms().getProducerTemplate();

        // send String
        template.sendBody(DIRECT_URI, "test");

        List<SourceRecord> poll1 = sourceTask.poll();
        assertEquals(1, poll1.size());
        assertEquals("test", poll1.get(0).value());
        assertEquals(Schema.Type.STRING, poll1.get(0).valueSchema().type());
        assertNull(poll1.get(0).key());
        assertNull(poll1.get(0).keySchema());

        // send boolean
        template.sendBody(DIRECT_URI, true);

        List<SourceRecord> poll2 = sourceTask.poll();
        assertEquals(1, poll2.size());
        assertTrue((boolean)poll2.get(0).value());
        assertEquals(Schema.Type.BOOLEAN, poll2.get(0).valueSchema().type());
        assertNull(poll2.get(0).key());
        assertNull(poll2.get(0).keySchema());

        // send long
        template.sendBody(DIRECT_URI, 1234L);

        List<SourceRecord> poll3 = sourceTask.poll();
        assertEquals(1, poll3.size());
        assertEquals(1234L, poll3.get(0).value());
        assertEquals(Schema.Type.INT64, poll3.get(0).valueSchema().type());
        assertNull(poll3.get(0).key());
        assertNull(poll3.get(0).keySchema());

        // send null
        template.sendBody(DIRECT_URI, null);

        List<SourceRecord> poll4 = sourceTask.poll();
        assertNull(poll4.get(0).key());
        assertNull(poll4.get(0).keySchema());
        assertNull(poll4.get(0).value());
        assertNull(poll4.get(0).valueSchema());

        sourceTask.stop();
    }

    @Test
    public void testUrlPrecedenceOnComponentProperty() throws InterruptedException {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, "timer:foo?period=10&repeatCount=2");
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "shouldNotBeUsed");
        props.put(CamelSourceTask.getCamelSourcePathConfigPrefix() + "timerName", "shouldNotBeUsed");
        props.put(CamelSourceTask.getCamelSourceEndpointConfigPrefix() + "repeatCount", "999");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sourceTask.getCms().getCamelContext().getEndpoints().stream()
                .filter(e -> e.getEndpointUri().startsWith("timer"))
                .forEach(e -> {
                    assertTrue(e.getEndpointUri().contains("foo"));
                    assertTrue(e.getEndpointUri().contains("period=10"));
                    assertTrue(e.getEndpointUri().contains("repeatCount=2"));
                });

        sourceTask.stop();
    }

    @Test
    public void testSourcePollingConsumerOptions() {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, "timer:foo?period=10&repeatCount=2");
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_QUEUE_SIZE_CONF, "10");
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_BLOCK_TIMEOUT_CONF, "10");
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_BLOCK_WHEN_FULL_CONF, "false");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sourceTask.getCms().getCamelContext().getEndpoints().stream()
                .filter(e -> e.getEndpointUri().startsWith("seda"))
                .forEach(e -> {
                    assertTrue(e.getEndpointUri().contains("end"));
                    assertTrue(e.getEndpointUri().contains("pollingConsumerQueueSize=10"));
                    assertTrue(e.getEndpointUri().contains("pollingConsumerBlockTimeout=10"));
                    assertTrue(e.getEndpointUri().contains("pollingConsumerBlockWhenFull=false"));
                });

        sourceTask.stop();
    }

    @Test
    public void testSourceUsingComponentProperties() {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "timer");
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MAX_BATCH_POLL_SIZE_CONF, "2");
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MAX_POLL_DURATION_CONF, "10");
        props.put(CamelSourceTask.getCamelSourcePathConfigPrefix() + "timerName", "foo");
        props.put(CamelSourceTask.getCamelSourceEndpointConfigPrefix() + "period", "10");
        props.put(CamelSourceTask.getCamelSourceEndpointConfigPrefix() + "repeatCount", "2");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        sourceTask.getCms().getCamelContext().getEndpoints().stream()
            .filter(e -> e.getEndpointUri().startsWith("timer"))
            .forEach(e -> {
                assertTrue(e.getEndpointUri().contains("foo"));
                assertTrue(e.getEndpointUri().contains("period=10"));
                assertTrue(e.getEndpointUri().contains("repeatCount=2"));
            });

        sourceTask.stop();
    }

    @Test
    public void testSourceBigDecimalHeader() {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "direct");
        props.put(CamelSourceTask.getCamelSourcePathConfigPrefix() + "name", "start");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);


        final ProducerTemplate template = sourceTask.getCms().getProducerTemplate();
        template.sendBodyAndHeader(DIRECT_URI, "test", "bigdecimal", new BigDecimal(1234567890));

        List<SourceRecord> results = sourceTask.poll();
        assertEquals(1, results.size());
        Header bigDecimalHeader = results.get(0).headers().allWithName(CamelSourceTask.HEADER_CAMEL_PREFIX + "bigdecimal").next();
        assertTrue(bigDecimalHeader.value() instanceof BigDecimal);
        assertEquals(Decimal.class.getName(), bigDecimalHeader.schema().name());
        assertEquals(Schema.Type.BYTES, bigDecimalHeader.schema().type());

        sourceTask.stop();
    }

    @Test
    public void testSourceByteArrayHeader() {
        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(
            CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME,
            CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "direct",
            CamelSourceTask.getCamelSourcePathConfigPrefix() + "name", "start"
        ));

        sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "test", "byteArray", new Byte[] {
            1, 2
        });

        try {
            List<SourceRecord> results = sourceTask.poll();
            assertThat(results).hasSize(1);

            Header header = results.get(0).headers().allWithName(CamelSourceTask.HEADER_CAMEL_PREFIX + "byteArray").next();

            assertThat(header.schema().type()).isEqualTo(Schema.Type.BYTES);
            assertThat(header.value()).isInstanceOfSatisfying(byte[].class, b -> {
                assertThat(b).contains(1, 2);
            });
        } finally {
            sourceTask.stop();
        }
    }
    
    @Test
    public void testSourceByteArrayProperty() {
        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(
            CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME,
            CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "direct",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_MAP_PROPERTIES_CONF, "false",
            CamelSourceTask.getCamelSourcePathConfigPrefix() + "name", "start"
        ));

        sourceTask.getCms().getProducerTemplate().sendBodyAndProperty(DIRECT_URI, "test", "byteArray", new Byte[] {
            1, 2
        });

        try {
            List<SourceRecord> results = sourceTask.poll();
            assertThat(results).hasSize(1);

            assertEquals(0, results.get(0).headers().size());
        } finally {
            sourceTask.stop();
        }
    }
    
    @Test
    public void testSourceByteArrayHeaderMapping() {
        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(
            CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME,
            CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "direct",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_MAP_HEADERS_CONF, "false",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_MAP_PROPERTIES_CONF, "false",
            CamelSourceTask.getCamelSourcePathConfigPrefix() + "name", "start"
        ));

        sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "test", "byteArray", new Byte[] {
            1, 2
        });

        try {
            List<SourceRecord> results = sourceTask.poll();
            assertThat(results).hasSize(1);

            assertEquals(0, results.get(0).headers().size());
        } finally {
            sourceTask.stop();
        }
    }

    @Test
    public void testSourceDateHeader() {
        final String key = "_key";
        final Date now = new Date();

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(
            CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME,
            CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "direct",
            CamelSourceTask.getCamelSourcePathConfigPrefix() + "name", "start"
        ));

        sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "test", key, now);

        try {
            List<SourceRecord> results = sourceTask.poll();
            assertThat(results).hasSize(1);

            Header header = results.get(0).headers().allWithName(CamelSourceTask.HEADER_CAMEL_PREFIX + key).next();

            assertThat(header.schema().type()).isEqualTo(Schema.Type.INT64);
            assertThat(header.value()).isInstanceOfSatisfying(Date.class, value -> {
                assertThat(value).isEqualTo(now);
            });
        } finally {
            sourceTask.stop();
        }
    }

    @Test
    public void testSourcePollingWithAggregationBySize() {
        final int size = 10;
        final int chunkSize = 5;

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(
            CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME,
            CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI,
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF, "#class:org.apache.camel.kafkaconnector.utils.StringJoinerAggregator",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF + ".delimiter", "|",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, chunkSize
        ));

        try {
            assertThat(sourceTask.getCms().getCamelContext().getRegistry().lookupByName(CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME))
                .isInstanceOf(StringJoinerAggregator.class)
                .hasFieldOrPropertyWithValue("delimiter", "|");

            for (int i = 0; i < size; i++) {
                sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI,  Integer.toString(i));
            }

            List<SourceRecord> records = sourceTask.poll();

            assertThat(records).hasSize(size / chunkSize);

            for (int i = 0; i < size / chunkSize; i++) {
                assertThat(records)
                .element(i)
                .hasFieldOrPropertyWithValue(
                    "value",
                    IntStream.range(i * chunkSize, (i * chunkSize) + chunkSize).mapToObj(Integer::toString).collect(Collectors.joining("|"))
                );
            }

        } finally {
            sourceTask.stop();
        }
    }

    @Test
    public void testSourcePollingWithAggregationBySizeAndTimeout() {
        final int size = 3;
        final int chunkSize = 2;
        final long chunkTimeout = 500L;

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(
            CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME,
            CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI,
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF, "#class:org.apache.camel.kafkaconnector.utils.StringJoinerAggregator",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF + ".delimiter", "|",
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_CONF, chunkTimeout,
            CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, chunkSize
        ));

        try {
            assertThat(sourceTask.getCms().getCamelContext().getRegistry().lookupByName(CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME))
                .isInstanceOf(StringJoinerAggregator.class)
                .hasFieldOrPropertyWithValue("delimiter", "|");

            for (int i = 0; i < size; i++) {
                sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, Integer.toString(i));
            }

            List<SourceRecord> records = new ArrayList<>();
            while (records.size() < 2) {
                records.addAll(sourceTask.poll());
            }

            assertThat(records).hasSize(2);
            assertThat(records).element(0).hasFieldOrPropertyWithValue("value", "0|1");
            assertThat(records).element(1).hasFieldOrPropertyWithValue("value", "2");
        } finally {
            sourceTask.stop();
        }
    }
    
    @Test
    public void testSourcePollingWithIdempotencyEnabledAndBody() {

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(mapOf(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME, CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI,
                               CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, true,
                               CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, "body"));

        try {

            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, "Test");
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, "Test1");
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, "Test");
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, "Test2");

            List<SourceRecord> records = sourceTask.poll();

            assertThat(records).hasSize(3);
            assertThat(records).element(0).hasFieldOrPropertyWithValue("value", "Test");
            assertThat(records).element(1).hasFieldOrPropertyWithValue("value", "Test1");
            assertThat(records).element(2).hasFieldOrPropertyWithValue("value", "Test2");
        } finally {
            sourceTask.stop();
        }
    }

    @Test
    public void testSourcePollingWithIdempotencyEnabledAndHeader() {

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask
            .start(mapOf(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME, CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI,
                         CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, true, CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF,
                         "header", CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_CONF, "headerIdempotency"));

        try {

            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "Test", "headerIdempotency", "Test");
            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "Test1", "headerIdempotency", "Test1");
            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "TestTest", "headerIdempotency", "Test");
            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "Test2", "headerIdempotency", "Test2");

            List<SourceRecord> records = sourceTask.poll();

            assertThat(records).hasSize(3);
            assertThat(records).element(0).hasFieldOrPropertyWithValue("value", "Test");
            assertThat(records).element(1).hasFieldOrPropertyWithValue("value", "Test1");
            assertThat(records).element(2).hasFieldOrPropertyWithValue("value", "Test2");
        } finally {
            sourceTask.stop();
        }
    }

    @Test
    public void testSourcePollingWithAggregationAndIdempotencyBySizeAndTimeout() {
        final int chunkSize = 2;

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask
            .start(mapOf(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME, CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI,
                         CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, true, CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF,
                         "body", CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF, "#class:org.apache.camel.kafkaconnector.utils.StringJoinerAggregator",
                         CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF + ".delimiter", "|", CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, chunkSize));

        try {
            assertThat(sourceTask.getCms().getCamelContext().getRegistry().lookupByName(CamelSourceConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME))
                .isInstanceOf(StringJoinerAggregator.class).hasFieldOrPropertyWithValue("delimiter", "|");

            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 0);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 1);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 2);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 3);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 0);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 1);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 3);
            sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI, 2);

            List<SourceRecord> records = sourceTask.poll();

            assertThat(records).hasSize(3);
            assertThat(records).element(0).hasFieldOrPropertyWithValue("value", "0|1");
            assertThat(records).element(1).hasFieldOrPropertyWithValue("value", "2|3");
            assertThat(records).element(2).hasFieldOrPropertyWithValue("value", "3|2");
        } finally {
            sourceTask.stop();
        }
    }
    
    @Test
    public void testSourcePollingWithIdempotencyEnabledAndHeaderExclusion() {

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask
            .start(mapOf(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME, CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI,
                         CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, true, CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF,
                         "header", CamelSourceConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_CONF, "headerIdempotency",
                         CamelSourceConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF, "headerIdempotency"));

        try {

            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "Test", "headerIdempotency", "Test");
            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "Test1", "headerIdempotency", "Test1");
            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "TestTest", "headerIdempotency", "Test");
            sourceTask.getCms().getProducerTemplate().sendBodyAndHeader(DIRECT_URI, "Test2", "headerIdempotency", "Test2");

            List<SourceRecord> records = sourceTask.poll();

            assertThat(records).hasSize(3);
            assertThat(records).element(0).hasFieldOrPropertyWithValue("value", "Test");
            assertThat(records).element(1).hasFieldOrPropertyWithValue("value", "Test1");
            assertThat(records).element(2).hasFieldOrPropertyWithValue("value", "Test2");
            assertFalse(records.get(0).headers().allWithName("headerIdempotency").hasNext());
            assertFalse(records.get(1).headers().allWithName("headerIdempotency").hasNext());
            assertFalse(records.get(2).headers().allWithName("headerIdempotency").hasNext());
        } finally {
            sourceTask.stop();
        }
    }

    @Test
    public void testContentLogLevelConfiguration() {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_CONTENT_LOG_LEVEL_CONF, "INFO");

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);
        assertEquals(LoggingLevel.INFO, sourceTask.getLoggingLevel());

        sourceTask.stop();
    }

    @Test
    public void testRequestReply() throws InterruptedException {
        Map<String, String> props = new HashMap<>();
        props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF, DIRECT_URI);

        CamelSourceTask sourceTask = new CamelSourceTask();
        sourceTask.start(props);

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                final ProducerTemplate template = sourceTask.getCms().getProducerTemplate();
                String result = template.requestBody(DIRECT_URI, "test", String.class);
                assertEquals("test", result);
            }
        });

        List<SourceRecord> poll = sourceTask.poll();
        assertEquals(1, poll.size());

        sourceTask.commitRecord(poll.get(0), null);

        sourceTask.stop();
        executor.shutdown();
    }
}