Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.camel.kafkaconnector.CamelSinkConnectorConfig

[saranyaeu2987]

I am trying to run camel-kafka-connector in my minikube and Strimzi but getting
Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.camel.kafkaconnector.CamelSinkConnectorConfig

Here is my image Docker file

FROM strimzi/kafka:0.18.0-kafka-2.5.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

Here is my plugins folder contains the following jars

When I see the pod logs, this is what I see

2020-05-28 16:17:11,313 INFO Registered loader: PluginClassLoader{pluginLocation=file:/opt/kafka/plugins/camel-kafka-connector-0.2.0.jar} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2020-05-28 16:17:11,313 INFO Added plugin 'org.apache.camel.kafkaconnector.CamelSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2020-05-28 16:17:11,313 INFO Added plugin 'org.apache.camel.kafkaconnector.CamelSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2020-05-28 16:17:11,313 INFO Added plugin 'org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2020-05-28 16:17:11,313 INFO Added plugin 'org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]

I dont see org.apache.camel.kafkaconnector.CamelSinkConnectorConfig getting loaded from camel-kafka-connector-0.2.0.jar
Which I feel would have caused the following error

Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.camel.kafkaconnector.CamelSinkConnectorConfig
	at org.apache.camel.kafkaconnector.CamelSinkConnector.config(CamelSinkConnector.java:67)
	at org.apache.kafka.connect.connector.Connector.validate(Connector.java:129)
...

**Full log from connector pod **
logs-from-my-connect-cluster-connect-in-my-connect-cluster-connect-8866c5d89-rd4zs.txt

Any suggestions??

[oscerd]

Can you please show your configuration file?

[saranyaeu2987]

@oscerd
Here is my kafkaconnect Strimzi deployment yaml file

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  image: selumalai/selumalai-s3-kafkaconnect #(Docker file definition in above comment where my-plugins contains all camel connector jars)
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  replicas: 1
  bootstrapServers: 34.212.160.12:9092
  externalConfiguration:
    volumes:
      - name: aws-credentials
        secret:
          secretName: aws-credentials
  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

Here is my KafkaConnector deployment file

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaConnector
metadata:
  name: s3-sink-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.camel.kafkaconnector.CamelSinkConnector
  tasksMax: 1
  config:
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.storage.StringConverter
    topics: my-replicated-topic
    camel.sink.url: aws-s3://selumalai-kafka-s3?keyName=${date:now:yyyyMMdd-HHmmssSSS}-${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: US_WEST_2

[valdar]

@saranyaeu2987 can you please try with this Docker file:

FROM strimzi/kafka:0.18.0-kafka-2.5.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/camel-aws-s3/
USER 1001

[saranyaeu2987]

Worked. Thanks a lot!!.

BTW, I have this
camel.sink.url: aws-s3://selumalai-kafka-s3?keyName=${date:now:yyyyMMdd-HHmmssSSS}-${exchangeId}

1. Why the variables are not resolved and the file is saved as ${date:now:yyyyMMdd-HHmmssSSS}-${exchangeId} ??

2. Is there a batch size property for S3 sink ( since I couldnt find any example for S3 sink https://github.com/apache/camel-kafka-connector/tree/master/examples)?
   something like this
   camel.sink.maxBatchPollSize = 1000

[oscerd]

The simple language is not resolved in the sink.url

[saranyaeu2987]

@valdar @oscerd

I want my CamelSinkConnector(s3) to consume in batches of 1000 messages or for interval of 10seconds.
How can I set this property in yaml file ?

[oscerd]

The S3 sink connector doens't provide a batch size option because it's mapped on the camel producer options actually. The producer doesn't work as a batch job.

[saranyaeu2987]

So, this property CamelBatchSize in https://camel.apache.org/manual/latest/batch-consumer.html
will not be applicable to s3sink connector?

[oscerd]

Yes, because you're mixing up the concepts: a consumer in Camel will be used when you're using a Source connector, so you'll have the consumer options in that case.
For example if you use the source connector you can set the

camel.source.endpoint.maxMessagesPerPoll

And define how much messages will be part of the batch consumed.

On the sink side you will be using the producer options.

Source options: https://camel.apache.org/camel-kafka-connector/latest/connectors/camel-aws-s3-kafka-source-connector.html
Sink options: https://camel.apache.org/camel-kafka-connector/latest/connectors/camel-aws-s3-kafka-sink-connector.html

You're using a sink connector (in Camel terms you're using a producer)

[saranyaeu2987]

Got it.
I wanted to add variable date in camel.sink.url. is there a way to add it.
without variable part, the file in S3 is getting overwritten.

Here is my yaml file for your reference

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaConnector
metadata:
  name: s3-sink-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.camel.kafkaconnector.CamelSinkConnector
  tasksMax: 1
  config:
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.storage.StringConverter
    topics: my-replicated-topic
    camel.sink.url: aws-s3://selumalai-kafka-s3?keyName=${date:now:yyyyMMdd-HHmmssSSS}-${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: US_WEST_2

[oscerd]

I don't think it's possible with simple language. So I guess you'll need to add something dynamic there.

[saranyaeu2987]

@oscerd
I see both ${date:now:yyyyMMdd} and exchangeId in this document (https://camel.apache.org/components/latest/languages/simple-language.html)

Why is it's not possible in simple language? Pardon me if I again mixup concepts

[oscerd]

It's not possible because you're declaring this stuff in a static file, simple language should be used in a real Camel route. The kafka configuration for connector is not dynamic, it's static.

[saranyaeu2987]

@oscerd @valdar

If that the case, how file variable is resolved?

NOT RESOLVED -->camel.sink.url: aws-s3://selumalai-kafka-s3?keyName=${date:now:yyyyMMdd-HHmmssSSS}-${exchangeId}. 
RESOLVED -->camel.component.aws-s3.accessKey: ${file:/opt/kafka/external-configuration/aws-credentials/aws-credentials.properties:aws_access_key_id}. 

=======================================================================

Basically, I wanted to preserve all Kafka topic data consumed via sinkconnector in S3 without overwriting.

Any suggestion on

1. How to prevent overwriting S3 file after consuming data from a kafka topic?
   OR
   How can I add dynamic part to url so that sink connector writes topics data to different files in s3
2. Is someway to write in batches in S3 ?

Current State: S3 file overwritten after every consumption
Desired State : Preserve every consumed data in S3. If batch consumption (say 1000 messages/file) available, it would be fantastic!!)