/*
 * 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.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.storage.SimpleHeaderConverter;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

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.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

class CamelSinkTaskTest {

    private static final String SEDA_URI = "seda:test";
    private static final String TOPIC_NAME = "my-topic";
    private static final long RECEIVE_TIMEOUT = 1_000;
    private static final String TOPIC_CONF = "topics";

    private CamelSinkTask sinkTask;

    @BeforeEach
    void setup() {
        sinkTask = new CamelSinkTask();
    }

    @AfterEach
    void tearDown() {
        sinkTask.stop();
    }


    @Test
    void testOnlyBody() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
    }

    @Test
    void testTopicsRegex() {
        Map<String, String> props = new HashMap<>();
        props.put("topics.regex", "topic1*");
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord("topic1", 1, null, "test", null, "camel", 42);
        SinkRecord record1 = new SinkRecord("topic12", 1, null, "test", null, "cameltopicregex", 42);
        records.add(record);
        records.add(record1);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
        Exchange exchange1 = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("cameltopicregex", exchange1.getMessage().getBody());
        assertEquals("test", exchange1.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
    }

    @Test
    void testBodyAndHeaders() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        BigDecimal myBigDecimal = new BigDecimal(1234567890);
        Schema myBigDecimalSchema = Decimal.schema(myBigDecimal.scale());
        //reproducing bigDecimal encoding by kafka connect
        BigDecimal kafkaBigDecimal = new BigDecimal("6.9203120E+787");
        Schema kafkaBigDecimalSchema = Decimal.schema(kafkaBigDecimal.scale());
        SimpleHeaderConverter shc = new SimpleHeaderConverter();
        byte[] persistedBytes = shc.fromConnectHeader("", "MyBigDecimal", kafkaBigDecimalSchema, kafkaBigDecimal);
        SchemaAndValue sav = shc.toConnectHeader("", "MyBigDecimal", persistedBytes);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().add(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBigDecimal", Decimal.fromLogical(myBigDecimalSchema, myBigDecimal), myBigDecimalSchema);
        record.headers().add(CamelSinkTask.HEADER_CAMEL_PREFIX + "KafkaBigDecimal", sav.value(), sav.schema());

        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
        assertEquals(myBigDecimal, exchange.getIn().getHeader("MyBigDecimal", BigDecimal.class));
        assertEquals(kafkaBigDecimal, exchange.getIn().getHeader("KafkaBigDecimal", BigDecimal.class));
    }
    
    @Test
    void testBodyAndHeadersExclusions() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF, "MyBoolean" + "|" + "MyShort");

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        BigDecimal myBigDecimal = new BigDecimal(1234567890);
        Schema schema = Decimal.schema(myBigDecimal.scale());

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().add(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBigDecimal", Decimal.fromLogical(schema, myBigDecimal), schema);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertNull(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertNull(exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
        assertEquals(myBigDecimal, exchange.getIn().getHeader("MyBigDecimal", BigDecimal.class));
    }
    
    @Test
    void testBodyAndHeadersExclusionsRegex() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF, "My*");

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        BigDecimal myBigDecimal = new BigDecimal(1234567890);
        Schema schema = Decimal.schema(myBigDecimal.scale());

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().add(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBigDecimal", Decimal.fromLogical(schema, myBigDecimal), schema);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertNull(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertNull(exchange.getIn().getHeader("MyByte", Byte.class));
        assertNull(exchange.getIn().getHeader("MyFloat", Float.class));
        assertNull(exchange.getIn().getHeader("MyShort", Short.class));
        assertNull(exchange.getIn().getHeader("MyDouble", Double.class));
        assertNull(exchange.getIn().getHeader("MyInteger"));
        assertNull(exchange.getIn().getHeader("MyLong", Long.class));
        assertNull(exchange.getIn().getHeader("MyBigDecimal", BigDecimal.class));
    }

    @Test
    void testBodyAndProperties() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyLong", myLong);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue((boolean) exchange.getProperties().get("MyBoolean"));
        assertEquals(myByte, (Byte) exchange.getProperties().get("MyByte"));
        assertEquals(myFloat, (Float) exchange.getProperties().get("MyFloat"));
        assertEquals(myShort, (Short) exchange.getProperties().get("MyShort"));
        assertEquals(myDouble, (Double) exchange.getProperties().get("MyDouble"));
        assertEquals(myInteger, exchange.getProperties().get("MyInteger"));
        assertEquals(myLong, (Long) exchange.getProperties().get("MyLong"));
    }

    @Test
    void testBodyAndPropertiesHeadersMixed() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue((boolean) exchange.getProperties().get("MyBoolean"));
        assertEquals(myByte, (Byte) exchange.getProperties().get("MyByte"));
        assertEquals(myFloat, (Float) exchange.getProperties().get("MyFloat"));
        assertEquals(myShort, (Short) exchange.getProperties().get("MyShort"));
        assertEquals(myDouble, (Double) exchange.getProperties().get("MyDouble"));
        assertEquals(myInteger, exchange.getProperties().get("MyInteger"));
        assertEquals(myLong, (Long) exchange.getProperties().get("MyLong"));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
    }

    @Test
    void testBodyAndHeadersMap() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        HashMap<String, String> map = new HashMap<String, String>();
        map.put("a", "a");
        HashMap<Integer, String> map1 = new HashMap<Integer, String>();
        map1.put(1, "a");
        HashMap<Integer, Integer> map2 = new HashMap<Integer, Integer>();
        map2.put(1, 1);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addMap(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyMap", map, SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA));
        record.headers().addMap(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyMap1", map1, SchemaBuilder.map(Schema.INT64_SCHEMA, Schema.STRING_SCHEMA));
        record.headers().addMap(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyMap2", map2, SchemaBuilder.map(Schema.INT64_SCHEMA, Schema.INT64_SCHEMA));
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
        assertEquals(map, exchange.getIn().getHeader("MyMap", Map.class));
        assertEquals(map1, exchange.getIn().getHeader("MyMap1", Map.class));
        assertEquals(map2, exchange.getIn().getHeader("MyMap2", Map.class));
    }

    @Test
    void testBodyAndPropertiesHeadersMapMixed() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        HashMap<String, String> map = new HashMap<String, String>();
        map.put("a", "a");
        HashMap<Integer, String> map1 = new HashMap<Integer, String>();
        map1.put(1, "a");
        HashMap<Integer, Integer> map2 = new HashMap<Integer, Integer>();
        map2.put(1, 1);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addMap(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyMap", map, SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA));
        record.headers().addMap(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyMap1", map1, SchemaBuilder.map(Schema.INT64_SCHEMA, Schema.STRING_SCHEMA));
        record.headers().addMap(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyMap2", map2, SchemaBuilder.map(Schema.INT64_SCHEMA, Schema.INT64_SCHEMA));
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addMap(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyMap", map, SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA));
        record.headers().addMap(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyMap1", map1, SchemaBuilder.map(Schema.INT64_SCHEMA, Schema.STRING_SCHEMA));
        record.headers().addMap(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyMap2", map2, SchemaBuilder.map(Schema.INT64_SCHEMA, Schema.INT64_SCHEMA));
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue((boolean) exchange.getProperties().get("MyBoolean"));
        assertEquals(myByte, (Byte) exchange.getProperties().get("MyByte"));
        assertEquals(myFloat, (Float) exchange.getProperties().get("MyFloat"));
        assertEquals(myShort, (Short) exchange.getProperties().get("MyShort"));
        assertEquals(myDouble, (Double) exchange.getProperties().get("MyDouble"));
        assertEquals(myInteger, exchange.getProperties().get("MyInteger"));
        assertEquals(myLong, (Long) exchange.getProperties().get("MyLong"));
        assertEquals(map, exchange.getProperties().get("MyMap"));
        assertEquals(map1, exchange.getProperties().get("MyMap1"));
        assertEquals(map2, exchange.getProperties().get("MyMap2"));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
        assertEquals(map, exchange.getIn().getHeader("MyMap", Map.class));
        assertEquals(map1, exchange.getIn().getHeader("MyMap1", Map.class));
        assertEquals(map2, exchange.getIn().getHeader("MyMap2", Map.class));
    }

    @Test
    void testBodyAndHeadersList() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        List<String> list = new ArrayList<String>();
        list.add("a");
        List<Integer> list1 = new ArrayList<Integer>();
        list1.add(1);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addList(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyList", list, SchemaBuilder.array(Schema.STRING_SCHEMA));
        record.headers().addList(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyList1", list1, SchemaBuilder.array(Schema.INT64_SCHEMA));
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
        assertEquals(list, exchange.getIn().getHeader("MyList", List.class));
        assertEquals(list1, exchange.getIn().getHeader("MyList1", List.class));
    }

    @Test
    void testBodyAndPropertiesHeadersListMixed() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");
        List<String> list = new ArrayList<String>();
        list.add("a");
        List<Integer> list1 = new ArrayList<Integer>();
        list1.add(1);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addList(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyList", list, SchemaBuilder.array(Schema.STRING_SCHEMA));
        record.headers().addList(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyList1", list1, SchemaBuilder.array(Schema.INT64_SCHEMA));
        record.headers().addList(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyList", list, SchemaBuilder.array(Schema.STRING_SCHEMA));
        record.headers().addList(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyList1", list1, SchemaBuilder.array(Schema.INT64_SCHEMA));
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertTrue((boolean) exchange.getProperties().get("MyBoolean"));
        assertEquals(myByte, (Byte) exchange.getProperties().get("MyByte"));
        assertEquals(myFloat, (Float) exchange.getProperties().get("MyFloat"));
        assertEquals(myShort, (Short) exchange.getProperties().get("MyShort"));
        assertEquals(myDouble, (Double) exchange.getProperties().get("MyDouble"));
        assertEquals(myInteger, exchange.getProperties().get("MyInteger"));
        assertEquals(myLong, (Long) exchange.getProperties().get("MyLong"));
        assertEquals(list, exchange.getProperties().get("MyList"));
        assertEquals(list1, exchange.getProperties().get("MyList1"));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
        assertEquals(list, exchange.getIn().getHeader("MyList", List.class));
        assertEquals(list1, exchange.getIn().getHeader("MyList1", List.class));
    }

    @Test
    void testUrlPrecedenceOnComponentProperty() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_COMPONENT_CONF, "shouldNotBeUsed");
        props.put(CamelSinkTask.getCamelSinkEndpointConfigPrefix() + "endpointProperty", "shouldNotBeUsed");
        props.put(CamelSinkTask.getCamelSinkPathConfigPrefix() + "pathChunk", "shouldNotBeUsed");

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
    }

    @Test
    void testOnlyBodyUsingComponentProperty() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_COMPONENT_CONF, "seda");
        props.put(CamelSinkTask.getCamelSinkEndpointConfigPrefix() + "bridgeErrorHandler", "true");
        props.put(CamelSinkTask.getCamelSinkPathConfigPrefix() + "name", "test");

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(1, sinkTask.getCms().getCamelContext().getEndpoints()
            .stream().filter(e -> e.getEndpointUri().equals("seda://test?bridgeErrorHandler=true")).count());
    }

    @Test
    void testOnlyBodyUsingMultipleComponentProperties() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_COMPONENT_CONF, "seda");
        props.put(CamelSinkTask.getCamelSinkEndpointConfigPrefix() + "bridgeErrorHandler", "true");
        props.put(CamelSinkTask.getCamelSinkEndpointConfigPrefix() + "size", "50");
        props.put(CamelSinkTask.getCamelSinkPathConfigPrefix() + "name", "test");

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));

        assertEquals(1, sinkTask.getCms().getCamelContext().getEndpoints()
            .stream().filter(e -> e.getEndpointUri().equals("seda://test?bridgeErrorHandler=true&size=50")).count());
    }
    
    @Test
    void testBodyAndPropertiesHeadersMixedWithoutPropertiesAndHeadersMapping() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_MAP_PROPERTIES_CONF, "false");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_MAP_HEADERS_CONF, "false");

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertFalse(exchange.getProperties().containsKey("MyBoolean"));
        assertFalse(exchange.getProperties().containsKey("MyByte"));
        assertFalse(exchange.getProperties().containsKey("MyFloat"));
        assertFalse(exchange.getProperties().containsKey("MyShort"));
        assertFalse(exchange.getProperties().containsKey("MyDouble"));
        assertFalse(exchange.getProperties().containsKey("MyInteger"));
        assertFalse(exchange.getProperties().containsKey("MyLong"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyBoolean"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyByte"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyFloat"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyShort"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyDouble"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyInteger"));
        assertFalse(exchange.getMessage().getHeaders().containsKey("MyLong"));
    }
    
    @Test
    void testBodyAndPropertiesHeadersMixedWithoutPropertiesMapping() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_MAP_PROPERTIES_CONF, "false");

        sinkTask.start(props);

        Byte myByte = new Byte("100");
        Float myFloat = new Float("100");
        Short myShort = new Short("100");
        Double myDouble = new Double("100");
        int myInteger = 100;
        Long myLong = new Long("100");

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addBoolean(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.PROPERTY_CAMEL_PREFIX + "MyLong", myLong);
        record.headers().addBoolean(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyBoolean", true);
        record.headers().addByte(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyByte", myByte);
        record.headers().addFloat(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyFloat", myFloat);
        record.headers().addShort(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyShort", myShort);
        record.headers().addDouble(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDouble", myDouble);
        record.headers().addInt(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyInteger", myInteger);
        record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyLong", myLong);
        records.add(record);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertFalse(exchange.getProperties().containsKey("MyBoolean"));
        assertFalse(exchange.getProperties().containsKey("MyByte"));
        assertFalse(exchange.getProperties().containsKey("MyFloat"));
        assertFalse(exchange.getProperties().containsKey("MyShort"));
        assertFalse(exchange.getProperties().containsKey("MyDouble"));
        assertFalse(exchange.getProperties().containsKey("MyInteger"));
        assertFalse(exchange.getProperties().containsKey("MyLong"));
        assertTrue(exchange.getIn().getHeader("MyBoolean", Boolean.class));
        assertEquals(myByte, exchange.getIn().getHeader("MyByte", Byte.class));
        assertEquals(myFloat, exchange.getIn().getHeader("MyFloat", Float.class));
        assertEquals(myShort, exchange.getIn().getHeader("MyShort", Short.class));
        assertEquals(myDouble, exchange.getIn().getHeader("MyDouble", Double.class));
        assertEquals(myInteger, exchange.getIn().getHeader("MyInteger"));
        assertEquals(myLong, exchange.getIn().getHeader("MyLong", Long.class));
    }

    @Test
    void testIfExchangeFailsShouldThrowConnectException() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        // we use a dummy component sink in order fail the exchange delivery
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_COMPONENT_CONF, "direct");
        props.put(CamelSinkTask.getCamelSinkPathConfigPrefix() + "name", "test");

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        records.add(record);

        assertThrows(ConnectException.class, () -> sinkTask.put(records));
    }

    @Test
    void testAggregationBody() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF, "#class:org.apache.camel.kafkaconnector.utils.SampleAggregator");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, "5");
        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record1 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel1", 42);
        SinkRecord record2 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel2", 42);
        SinkRecord record3 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel3", 42);
        SinkRecord record4 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel4", 42);
        SinkRecord record5 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel5", 42);
        records.add(record);
        records.add(record1);
        records.add(record2);
        records.add(record3);
        records.add(record4);
        records.add(record5);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel camel1 camel2 camel3 camel4", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
    }

    @Test
    void testAggregationBodyAndTimeout() throws InterruptedException {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF, "#class:org.apache.camel.kafkaconnector.utils.SampleAggregator");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, "5");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_CONF, "100");
        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record1 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel1", 42);
        SinkRecord record2 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel2", 42);
        SinkRecord record3 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel3", 42);
        SinkRecord record4 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel4", 42);
        SinkRecord record5 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel5", 42);
        records.add(record);
        records.add(record1);
        records.add(record2);
        records.add(record3);
        records.add(record4);
        records.add(record5);
        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel camel1 camel2 camel3 camel4", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
    }
    
    @Test
    void testAggregationWithIdempotencyBodyAndTimeout() throws InterruptedException {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF, "#class:org.apache.camel.kafkaconnector.utils.SampleAggregator");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, "5");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_CONF, "100");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, "true");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, "body");
        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record1 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel1", 42);
        SinkRecord record2 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel2", 42);
        SinkRecord record3 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel3", 42);
        SinkRecord record4 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel4", 42);
        SinkRecord record5 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record6 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel1", 42);
        SinkRecord record7 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel2", 42);
        SinkRecord record8 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel3", 42);
        SinkRecord record9 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel4", 42);
        SinkRecord record10 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel5", 42);
        SinkRecord record11 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel6", 42);
        SinkRecord record12 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel7", 42);
        SinkRecord record13 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel8", 42);
        SinkRecord record14 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel9", 42);
        records.add(record);
        records.add(record1);
        records.add(record2);
        records.add(record3);
        records.add(record4);
        records.add(record5);
        records.add(record6);
        records.add(record7);
        records.add(record8);
        records.add(record9);
        records.add(record10);
        records.add(record11);
        records.add(record12);
        records.add(record13);
        records.add(record14);

        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel camel1 camel2 camel3 camel4", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
        
        exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel5 camel6 camel7 camel8 camel9", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
    }
    
    @Test
    void testWithIdempotency() throws InterruptedException {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, "true");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, "body");
        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record1 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record2 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record3 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record4 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel1", 42);
        SinkRecord record5 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record6 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record7 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record8 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record9 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel2", 42);
        SinkRecord record10 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record11 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record12 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record13 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        SinkRecord record14 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        records.add(record);
        records.add(record1);
        records.add(record2);
        records.add(record3);
        records.add(record4);
        records.add(record5);
        records.add(record6);
        records.add(record7);
        records.add(record8);
        records.add(record9);
        records.add(record10);
        records.add(record11);
        records.add(record12);
        records.add(record13);
        records.add(record14);

        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
        
        exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel1", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
        
        exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel2", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
    }
    
    @Test
    void testWithIdempotencyAndHeader() throws InterruptedException {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_ENABLED_CONF, "true");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_TYPE_CONF, "header");
        props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_IDEMPOTENCY_EXPRESSION_HEADER_CONF, "headerIdempotency");
        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<SinkRecord>();
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().add("CamelHeader.headerIdempotency", new SchemaAndValue(Schema.STRING_SCHEMA, "Test"));
        SinkRecord record1 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record1.headers().add("CamelHeader.headerIdempotency", new SchemaAndValue(Schema.STRING_SCHEMA, "Test"));
        SinkRecord record2 = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel1", 42);
        record2.headers().add("CamelHeader.headerIdempotency", new SchemaAndValue(Schema.STRING_SCHEMA, "Test1"));

        records.add(record);
        records.add(record1);
        records.add(record2);

        sinkTask.put(records);

        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
        
        exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertEquals("camel1", exchange.getMessage().getBody());
        assertEquals("test", exchange.getMessage().getHeaders().get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
        assertEquals(LoggingLevel.OFF.toString(), sinkTask.getCamelSinkConnectorConfig(props)
            .getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
    }

    @Test
    void testSecretRaw() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put("camel.sink.endpoint.secretKey", "se+ret");
        props.put("camel.sink.endpoint.accessKey", "MoreSe+ret$");
        props.put("camel.sink.endpoint.queueNameOrArn", "test");
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_COMPONENT_CONF, "aws2-sqs");

        sinkTask.start(props);
    }

    @Test
    void testSecretRawReference() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put("camel.sink.endpoint.secretKey", "#bean:mySecretKey");
        props.put("camel.sink.endpoint.accessKey", "#property:myAccessKey");
        props.put("camel.sink.endpoint.queueNameOrArn", "test");
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_COMPONENT_CONF, "aws2-sqs");
        props.put("myAccessKey", "MoreSe+ret$");

        sinkTask.start(props);
    }

    @Test
    void testBodyAndDateHeader() {
        final Date now = new Date();

        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);

        sinkTask.start(props);

        List<SinkRecord> records = new ArrayList<>();

        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 42);
        record.headers().addTimestamp(CamelSinkTask.HEADER_CAMEL_PREFIX + "MyDate", now);
        records.add(record);

        sinkTask.put(records);

        Exchange exchange = sinkTask.getCms().getConsumerTemplate().receive(SEDA_URI, RECEIVE_TIMEOUT);

        assertThat(exchange.getIn().getHeader("MyDate")).isInstanceOfSatisfying(Date.class, value -> {
            assertThat(value).isEqualTo(now);
        });
    }

    @Test
    void testContentLogLevelConfiguration() {
        Map<String, String> props = new HashMap<>();
        props.put(TOPIC_CONF, TOPIC_NAME);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        props.put(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF, "INFO");

        sinkTask.start(props);
        assertEquals(LoggingLevel.INFO, sinkTask.getLoggingLevel());
    }

    @Test
    void testThatSchemalessHeaderIsBeingMappedToExchange() {
        // given sink task
        Map<String, String> properties = new HashMap<>();
        properties.put(TOPIC_CONF, TOPIC_NAME);
        properties.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
        sinkTask.start(properties);

        // and source record
        String headerName = "test-header";
        Long headerValue = 1234L;
        SinkRecord record = new SinkRecord(TOPIC_NAME, 1, null, "test", null, "camel", 0);
        record.headers().add(CamelSinkTask.HEADER_CAMEL_PREFIX + headerName, headerValue, null);

        // when
        sinkTask.put(Collections.singleton(record));

        // then
        ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
        Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
        assertThat(exchange.getIn().getHeader(headerName)).isEqualTo(headerValue);
    }
}
