Merge branch 'master' into JAVA-42046

Author: panos-kakos

apache-kafka-3/src/main/java/com/baeldung/kafka/KafkaAddPartitionExample.java

@@ -0,0 +1,23 @@
+package com.baeldung.kafka;
+
+import java.util.Collections;
+import java.util.Properties;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.NewPartitions;
+
+public class KafkaAddPartitionExample {
+
+    public static void main(String[] args) {
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "localhost:9092");
+        try (AdminClient adminClient = AdminClient.create(props)) {
+            adminClient.createPartitions(Collections.singletonMap("my-topic", NewPartitions.increaseTo(3)))
+                .all()
+                .get();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/avro-sink-config.json

@@ -0,0 +1,14 @@
+{
+    "name": "avro-elasticsearch-sink",
+    "config": {
+      "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+      "tasks.max": "1",
+      "topics": "avro_logs",
+      "connection.url": "http://elasticsearch:9200",
+      "key.ignore": "true",
+      "schema.ignore": "false",
+      "value.converter": "io.confluent.connect.avro.AvroConverter",
+      "value.converter.schema.registry.url": "http://schema-registry:8081"
+    }
+  }
+  

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/dlq-sink-config.json

@@ -0,0 +1,20 @@
+{
+    "name": "dlq-elasticsearch-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "failing-topic",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "true",
+        "schema.ignore": "true",
+        
+        "behavior.on.malformed.documents": "ignore",
+        "errors.tolerance": "all",
+        "errors.log.enable": "true",
+        "errors.log.include.messages": "true",
+        
+        "errors.deadletterqueue.topic.name": "dlq-logs",
+        "errors.deadletterqueue.topic.replication.factor": "1",
+        "errors.deadletterqueue.context.headers.enable": "true"
+    }
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/error-handling-sink-config.json

@@ -0,0 +1,16 @@
+{
+    "name": "error-handling-elasticsearch-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "test-error-handling",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "true",
+        "schema.ignore": "true",
+        "behavior.on.malformed.documents": "warn",
+        "behavior.on.error": "LOG",
+        "errors.tolerance": "all",
+        "errors.log.enable": "true",
+        "errors.log.include.messages": "true"
+    }
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/flush-optimization-sink.json

@@ -0,0 +1,23 @@
+{
+    "name": "flush-optimization-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "logs",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "true",
+        "schema.ignore": "true",
+        
+        "flush.size": "500",
+        "linger.ms": "1000",
+        "batch.size": "200",
+        "max.buffered.records": "10000",
+        "max.retries": "10",
+        "retry.backoff.ms": "500",
+        "max.in.flight.requests": "10",
+        "behavior.on.malformed.documents": "warn",
+        "write.method": "bulk",
+        "request.timeout.ms": "30000",
+        "connection.timeout.ms": "10000"
+    }
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/message-id-sink-config.json

@@ -0,0 +1,19 @@
+{
+    "name": "message-id-elasticsearch-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "message_id_logs",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "false",
+        "schema.ignore": "true",
+        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+        "value.converter.schemas.enable": "false",
+        "transforms": "InsertKey,ExtractId",
+        "transforms.InsertKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
+        "transforms.InsertKey.fields": "user_id",
+        "transforms.ExtractId.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
+        "transforms.ExtractId.field": "user_id"
+    }
+}
+  

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/partition-key-sink-config.json

@@ -0,0 +1,12 @@
+{
+    "name": "partition-key-elasticsearch-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "partition_key_logs",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "false",
+        "schema.ignore": "true",
+        "key.converter": "org.apache.kafka.connect.storage.StringConverter"
+    }
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/test-connector.json

@@ -0,0 +1,12 @@
+{
+    "name": "elasticsearch-sink-connector-test",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "logs",
+        "connection.url": "http://elasticsearch:9200",
+        "type.name": "_doc",
+        "key.ignore": "true",
+        "schema.ignore": "true"
+    }
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/timestamp-index-sink.json

@@ -0,0 +1,20 @@
+{
+    "name": "timestamp-index-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "test",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "true",
+        "schema.ignore": "true",
+        "transforms": "TimestampConverter, TimestampRouter",
+        "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
+        "transforms.TimestampConverter.field": "timestamp",
+        "transforms.TimestampConverter.target.type": "string",
+        "transforms.TimestampConverter.format": "yyyy-MM-dd'T'HH:mm:ssX",
+        "transforms.TimestampRouter.type": "org.apache.kafka.connect.transforms.TimestampRouter",
+        "transforms.TimestampRouter.topic.format": "${topic}-${timestamp}",
+        "transforms.TimestampRouter.timestamp.format": "yyyy-MM-dd",
+        "flush.synchronously": "true"
+    }
+}

apache-kafka-3/src/main/resources/kafka-connector-elasticsearch/timestamp-transform-sink.json

@@ -0,0 +1,17 @@
+{
+    "name": "timestamp-transform-sink",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "epoch_logs",
+        "connection.url": "http://elasticsearch:9200",
+        "key.ignore": "true",
+        "schema.ignore": "true",
+        "transforms": "TimestampConverter",
+        "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
+        "transforms.TimestampConverter.field": "timestamp",
+        "transforms.TimestampConverter.target.type": "string",
+        "transforms.TimestampConverter.format": "yyyy-MM-dd'T'HH:mm:ssZ"
+    }
+}
+