Adding kafka connector to read data and push to ELK

docker-compose.yml

@@ -54,7 +54,10 @@ services:
       SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
 
   connect:
-    image: cnfldemos/cp-server-connect-datagen:0.4.0-6.1.0
+    #image: cnfldemos/cp-server-connect-datagen:0.4.0-6.1.0
+    build:
+      context: ./docker
+      dockerfile: DockerfileConnect
     hostname: connect
     container_name: connect
     depends_on:
@@ -176,3 +179,37 @@ services:
       KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'
       KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
       KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
+
+  postgressdb:
+    image: "postgres:11"
+    container_name: "postgress_kafka"
+    environment:
+      POSTGRES_HOST_AUTH_METHOD: "trust"
+    ports:
+      - "54320:5432"
+
+  es01:
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.12.0
+    container_name: es01
+    environment:
+      - bootstrap.memory_lock=true
+      - discovery.type=single-node
+      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    ports:
+      - 9200:9200
+      - 9300:9300
+
+  kib01:
+    image: docker.elastic.co/kibana/kibana:7.12.0
+    depends_on:
+      - es01
+    container_name: kib01
+    ports:
+      - 5601:5601
+    environment:
+      ELASTICSEARCH_URL: http://es01:9200
+      ELASTICSEARCH_HOSTS: '["http://es01:9200"]'

docker/DockerfileConnect

@@ -0,0 +1,6 @@
+FROM confluentinc/cp-kafka-connect:6.1.0
+
+ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"
+
+RUN  confluent-hub install --no-prompt confluentinc/kafka-connect-datagen:latest
+RUN  confluent-hub install --no-prompt confluentinc/kafka-connect-elasticsearch:latest

kafka-connect/etf-2-elk.json

@@ -0,0 +1,19 @@
+{
+    "name": "etf-simple-2-elk",
+    "config": {
+        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
+        "tasks.max": "1",
+        "topics": "etf_dummy_data",
+        "connection.url": "http://es01:9200",
+        "type.name": "_doc",
+        "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
+        "value.converter.schemas.enable": "false",
+        "value.converter.schema.registry.url": "http://schema-registry:8081",
+        "key.converter":"org.apache.kafka.connect.storage.StringConverter",
+        "key.converter.schemas.enable" : "false",
+        "schema.ignore": "true",
+        "key.ignore": "true",
+        "auto.create.indices.at.start":"true",
+        "write.method":"insert"
+    }
+}