Directions on how to connect strimzi kafka with the Elasticsearch

[shobha-my]

Hello,

I'm new to Kafka, I wanted directions on how to connect Kafka with the Elasticsearch. I have data in one kafka topic, want to push that to Elasticsearch cluster.

I know I need to use Kafka Connect, can't find easy straightforward articles to do this usecase easily.

Any help will be appreciated.

[scholzj]

Not an expert on this ... but I used it in this demo: https://github.com/scholzj/building-cloud-native-logging-pipelines-on-top-of-apache-kafka. I don't think I can help with anything more than what is covered int he files there, but maybe it helps.

[shobha-my]

@scholzj

I can't get it working, initially was getting the docker unauthorized error (I'm using same credentials for CI/CD pipeline added into the namespace correctly), now I'm getting agent not found error for the pod:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: odin-connect
  labels:
    strimzi.io/cluster: odin-prod-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # Kafka Connects internal topics
      - resource:
          type: group
          name: connect-cluster
        operation: Read
      - resource:
          type: topic
          name: connect-cluster-configs
        operation: Read
      - resource:
          type: topic
          name: connect-cluster-configs
        operation: Describe
      - resource:
          type: topic
          name: connect-cluster-configs
        operation: Write
      - resource:
          type: topic
          name: connect-cluster-configs
        operation: Describe
      - resource:
          type: topic
          name: connect-cluster-configs
        operation: Create
      - resource:
          type: topic
          name: connect-cluster-status
        operation: Read
      - resource:
          type: topic
          name: connect-cluster-status
        operation: Describe
      - resource:
          type: topic
          name: connect-cluster-status
        operation: Write
      - resource:
          type: topic
          name: connect-cluster-status
        operation: Describe
      - resource:
          type: topic
          name: connect-cluster-status
        operation: Create
      - resource:
          type: topic
          name: connect-cluster-offsets
        operation: Read
      - resource:
          type: topic
          name: connect-cluster-offsets
        operation: Describe
      - resource:
          type: topic
          name: connect-cluster-offsets
        operation: Write
      - resource:
          type: topic
          name: connect-cluster-offsets
        operation: Describe
      - resource:
          type: topic
          name: connect-cluster-offsets
        operation: Create
      # Connector topics: ElasticSearch Sink
      - resource:
          type: topic
          name: odin-ips
        operation: Describe
      - resource:
          type: topic
          name: odin-ips
        operation: Read
      - resource:
          type: group
          name: connect-elasticsearch-connector
        operation: Read
---

###################################################################################
# Uncomment the secret and update it with your own container repository credentials
###################################################################################

apiVersion: v1
kind: Secret
metadata:
  name: docker-credentials
type: kubernetes.io/dockerconfigjson
data:
  .dockerconfigjson: {base64_string}

---

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: odin-connect
  labels:
    app: odin-connect
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 1
  resources:
    requests:
      memory: 4Gi
      cpu: 4
    limits:
      memory: 8Gi
      cpu: 8
  logging:
    type: inline
    loggers:
      log4j.rootLogger: "INFO"
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  bootstrapServers: odin-prod-cluster-kafka-bootstrap.kafka:9092
  config:
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.storage.StringConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false
    config.providers: env
    config.providers.env.class: io.strimzi.kafka.EnvVarConfigProvider
  tls:
    trustedCertificates:
      - secretName: odin-prod-cluster-cluster-ca-cert
        certificate: ca.crt
  authentication:
    type: tls
    certificateAndKey:
      secretName: odin-connect
      certificate: user.crt
      key: user.key
  build:
    output:
      type: docker
      image: index.docker.io/v1/{org_name}/kafka-connect-build:latest
      pushSecret: docker-credentials
    plugins:
      - name: camel-elasticsearch
        artifacts:
          - type: tgz
            url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-elasticsearch-rest-kafka-connector/0.11.5/camel-elasticsearch-rest-kafka-connector-0.11.5-package.tar.gz
            sha512sum: edff772da41f3c5a007cf97477d31eeb2e3c6ae4bef728da0b5a9c768de47d2e4c55cd7eeb5b8af3a94917a14e1a0821e05cafedf3d9e65cea67e4a4d9ded133
          - type: jar
            url: https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-api/2.17.2/log4j-api-2.17.2.jar
            sha512sum: 8f986171ee6ca94ba8eacfc83b5b45bb87221c176076eb152715fcb4e1f351c4e9f88d56420fc02ee166ebe0b896fb52eda21355dec49302ab3e83a56b245d04

Any help will be appreciated.

[scholzj]

Your image format seems wrong. You should use a proper one (the same you would use to pull the image or as used in our examples). But without any logs, it is hard to say what the actual issue is.