Kafka Connect Failing Because of Liveliness and Readiness Probes

[stuartm21]

Hi Team,

I installed Kafka Cluster previously and now I am focusing on KafkaConnect and KafkaConnector. I used the following documentation and build the camel aws s3 sink kafka connecter, based on that I made a docker image and I pushed the image to the docker hub.

https://ibm-cloud-architecture.github.io/refarch-eda/scenarios/connect-s3/
https://developers.redhat.com/blog/2020/02/14/using-secrets-in-apache-kafka-connect-configuration/
Camel: https://github.com/apache/camel-kafka-connector

Dockerfile

FROM quay.io/strimzi/kafka:0.28.0-kafka-3.1.0
USER root:root
COPY camel-aws-s3-sink-kafka-connector-1.1.0-SNAPSHOT.jar /opt/kafka/plugins/
COPY camel-aws-s3-source-kafka-connector-1.1.0-SNAPSHOT.jar /opt/kafka/plugins/
USER 1001

KafkaConnect

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: strimzi-kafka
  namespace: kafka
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.1.0
  replicas: 1
  image: my-repo/camel-aws-s3-kafka-connector:v1.0
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  bootstrapServers: kafka-bootstrap.myhost.com:9094
  externalConfiguration:
    volumes:
      - name: aws-credentials
        secret:
          secretName: aws-credentials
  tls:
    trustedCertificates:
      - secretName: strimzi-cluster-ca-cert
        certificate: ca.crt
  config:
    config.providers: file
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false

KafkaConnector

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: s3-sink-connector
  namespace: kafka
  labels:
    strimzi.io/cluster: "strimzi"
spec:
  class: org.apache.camel.kafkaconnector.CamelSinkConnector
  tasksMax: 2
  config:
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.storage.StringConverter
    topics: kafka-topic
    camel.sink.url: aws-s3://kafka-connect-s3?keyName=${date:now:yyyyMMdd}/${exchangeId}
    camel.sink.maxPollDuration: 10000
    camel.component.aws-s3.configuration.autocloseBody: false
    camel.component.aws-s3.accessKey: ${file:/opt/kafka/external-configuration/aws-credentials/aws-credentials.properties:aws_access_key_id}
    camel.component.aws-s3.secretKey: ${file:/opt/kafka/external-configuration/aws-credentials/aws-credentials.properties:aws_secret_access_key}
    camel.component.aws-s3.region: CA_CENTRAL_1

But the problem is, the kafka connect pod is failing and the pod describe is here:

Events:
  Type     Reason     Age                   From               Message
  ----     ------     ----                  ----               -------
  Normal   Scheduled  30m                   default-scheduler  Successfully assigned kafka/strimzi-kafka-connect-5b8454c98d-v9n8c to node01
  Normal   Created    25m (x4 over 30m)     kubelet            Created container strimzi-kafka-connect
  Normal   Started    25m (x4 over 29m)     kubelet            Started container strimzi-kafka-connect
  Warning  Unhealthy  24m (x2 over 27m)     kubelet            Readiness probe failed: Get "http://10.244.7.251:8083/": dial tcp 10.244.7.251:8083: connect: connection refused
  Normal   Pulled     24m (x5 over 30m)     kubelet            Container image "my-repo/camel-aws-s3-kafka-connector:v1.0" already present on machine
  Warning  BackOff    9m58s (x56 over 27m)  kubelet            Back-off restarting failed container
  Warning  Unhealthy  4m40s (x7 over 27m)   kubelet            Liveness probe failed: Get "http://10.244.7.251:8083/": dial tcp 10.244.7.251:8083: connect: connection refused

As a testing purpose, I commented the tls like below inside the kafka connect, but that also didn't worked.

  #tls:
    #trustedCertificates:
      #- secretName: strimzi-cluster-ca-cert
        #certificate: ca.crt

We have not configured the liveness and readiness probe anywhere for this. Then why the pod is failing here?

Also I have another doubt, I have multiple kafka topics, so how can archive all topic to S3 bucket using the above connect yaml config?

strimzi-kafka-operator: 0.28.0
Kafka cluster version: 3.1.0

[scholzj]

You will have to check the different logs to find out what the problem is - fromt he operator, from Kafka Connect etc.

[scholzj]

I'm sorry, but this is not some paid support hotline which guarantees 1 hour responses. This is an open source community where people are answering your question often in their own time, during their evenings etc. If they know and have time, they might try to help. But I do not think pinging someone every 3 hours to solve your errors is the right approach. People might also not have time to share the exact docs or create your configuration YAML files for you. So you should also expect that you have to learn the things your self - study the docs, examples, demos etc.

The IBM guide you followed seems very old. The KafkaConnectS2I resource is not even supported anymore. The Camel 0.1.0 connector seems to be also old. So I'm not sure what exactly you did from it or not. The error says that you didn't added the connector - so you should doublecheck that you really did it, that it is in the container inside the pod etc.

[stuartm21]

@scholzj I'm sorry, I know this is not a paid support hotline or anything. Please try to understand my situation that's why I asked for updates again and again. I am working under high pressure and I am facing these issues for the last one week.

Our platform is K8s not Opensfift so I know that KafkaConnectS2I will not support for our infra. I am sure that I added many artifacts for resolving the issue. Please see the available plugins:

aws-java-sdk-core-1.12.173.jar
aws-java-sdk-s3-1.12.173.jar
camel-api-3.15.0.jar
camel-aws-s3-kafka-connector-0.8.0.jar
camel-aws-s3-sink-kafka-connector-1.0.0.jar
camel-aws-s3-source-kafka-connector-1.0.0.jar
camel-aws2-s3-3.15.0.jar
camel-base-3.15.0.jar
camel-core-3.15.0.jar
camel-kafka-connector-1.0.0.jar
camel-kamelets-0.7.1.jar
camel-main-3.15.0.jar
camel-util-3.15.0.jar
camel-util-json-3.15.0.jar
log4j-api-2.17.2.jar
log4j-core-2.17.2.jar
log4j-jcl-2.17.2.jar
log4j-slf4j-impl-2.17.2.jar

And many of the above artifacts I downloaded from the following link, I don't anything I missed to add to my connect. If anything please let me know or where we can found that?

https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/

Now I am getting the following logs from the kafkaconnect pods after I create the KafkaConnector:

2022-03-11 09:33:56,183 ERROR [s3-sink-connector|task-0] WorkerSinkTask{id=s3-sink-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-s3-sink-connector-0]
java.lang.NoClassDefFoundError: org/apache/camel/impl/DefaultCamelContext
  at org.apache.camel.kafkaconnector.CamelSinkTask.start(CamelSinkTask.java:118)
  at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:312)
  at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186)
  at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)
  at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
  at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.ClassNotFoundException: org.apache.camel.impl.DefaultCamelContext
  at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:476)
  at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:589)
  at org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass(PluginClassLoader.java:103)
  at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
  ... 9 more
2022-03-11 09:33:56,185 INFO [s3-sink-connector|task-0] Stopping CamelSinkTask connector task (org.apache.camel.kafkaconnector.CamelSinkTask) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,185 WARN [s3-sink-connector|task-0] A fatal exception may have occurred and the Camel main was not created (org.apache.camel.kafkaconnector.CamelSinkTask) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,185 INFO [s3-sink-connector|task-0] CamelSinkTask connector task stopped (org.apache.camel.kafkaconnector.CamelSinkTask) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,186 INFO [s3-sink-connector|task-0] [Consumer clientId=connector-consumer-s3-sink-connector-0, groupId=connect-s3-sink-connector] Resetting generation due to: consumer pro-actively leaving the group (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,186 INFO [s3-sink-connector|task-0] [Consumer clientId=connector-consumer-s3-sink-connector-0, groupId=connect-s3-sink-connector] Request joining group due to: consumer pro-actively leaving the group (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,186 INFO [s3-sink-connector|task-0] Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,186 INFO [s3-sink-connector|task-0] Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,186 INFO [s3-sink-connector|task-0] Metrics reporters closed (org.apache.kafka.common.metrics.Metrics) [task-thread-s3-sink-connector-0]
2022-03-11 09:33:56,188 INFO [s3-sink-connector|task-0] App info kafka.consumer for connector-consumer-s3-sink-connector-0 unregistered (org.apache.kafka.common.utils.AppInfoParser) [task-thread-s3-sink-connector-0]

I am new to kafka and I am tring to learn this. I have checked the strimzi official docs but the AWS s3 sink-related docs are not found anywhere.

[scholzj]

Kafka Connect supports hundreds of connectors. So no, Strimzi does not document how to use every single one of them. I have no clue what the issue is apart from the obvious error it says -> that some classes seem to be missing.

[stuartm21]

@scholzj yes.. some classes missing. That's why I am also stuck here. I don't know how to proceed further..

[cstmgl]

how are you injecting your jars into the docker?
meaning make sure your camel and other connectors are either fat jars (for example snowflake connector is) or they have all the necessary dependencies so that you can pull everything that is required at run time.