Strimzi-kafka-operator: [Bug] io.confluent.connect.avro.AvroConverter does not work as a key/value converter in KafkaConnectors

[timkalanai]

Describe the bug
This has got to be either a known bug, or I'm doing something stupid. But I'm trying to use io.confluent.connect.avro.AvroConverter for key and value (de)serialization. I've tried to use it for a few different Kafka Connect connectors, but to simulate it without much complexity, I used the s3 connector, and I can get it to mess up every time.

I've tried downloading this and putting it in my plugins directory, but it still doesn't seem to work: https://www.confluent.io/hub/confluentinc/kafka-connect-avro-converter

I started with a fairly straightforward configuration that worked. I copied the "kafka-connect-s3" directory from the confluent platform 5.5 (is there possibly a compatibility issue here??) directory. Also copied "kafka-connect-storage-common" (there's a dependency there...)

Everything seems to generally work pretty well until I try to use the AvroConverter. Looking in "kafka-connect-storage-common" there this jar: kafka-connect-avro-converter-5.5.0.jar which should be all I need.... But all I get is this:

    Tasks:
      Id:     0
      State:  FAILED
      Trace:  java.lang.NoClassDefFoundError: io/confluent/connect/avro/AvroConverterConfig
              at io.confluent.connect.avro.AvroConverter.configure(AvroConverter.java:64)
              at org.apache.kafka.connect.runtime.isolation.Plugins.newConverter(Plugins.java:266)
              at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:417)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:873)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1600(DistributedHerder.java:111)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:888)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:884)
              at java.util.concurrent.FutureTask.run(FutureTask.java:266)
              at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
              at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
              at java.lang.Thread.run(Thread.java:748)

In messing around with this configuration with other connectors, I've been abled to get: java.lang.NoClassDefFoundError for AbstractConfig as well sometimes... and then add those confluent common jars, then it goes back to not NoClassDefFoundError for AvroConverterConfig.

Something must be going on that I'm not seeing here.

Thanks!

To Reproduce
I use the following KafkaConnector configuration:

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaConnector
metadata:
  name: kafka-connector-s3-avro
  labels:
    strimzi.io/cluster: kafkaconnect-cluster
spec:
  class: io.confluent.connect.s3.S3SinkConnector
  tasksMax: 1
  config:
    format.class: io.confluent.connect.s3.format.json.JsonFormat
    s3.compression.type: gzip
    partitioner.class: io.confluent.connect.storage.partitioner.HourlyPartitioner
    topics: avrokafkamessagestopic
    s3.region: us-east-2
    s3.bucket.name: avrokafkamessages
    flush.size: 1
    storage.class: io.confluent.connect.s3.storage.S3Storage
    locale: en-US
    timezone: UTC
    schemas.enable: false
    key.converter: io.confluent.connect.avro.AvroConverter
    key.converter.schema.registry.url: http://schema-registry-release-cp-schema-registry:8081
    value.converter: io.confluent.connect.avro.AvroConverter
    value.converter.schema.registry.url: http://schema-registry-release-cp-schema-registry:8081

I know that what I'm doing isn't supported out of the box, so I've followed the many tutorials on how to create your own kafkaconnect image here's my docker file:

FROM strimzi/kafka-connect:0.11.4-kafka-2.1.0
USER root:root
COPY ./connect-plugins/ /opt/kafka/plugins/
USER 1001

I add the following directories to my ./connect-plugins directory:
kafka-connect-s3
kafka-connect-storage-common

from confluent 5.5 platform: share/java

And my KafkaConnect configuration:

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: kafkaconnect-cluster
  annotations:
#  # use-connector-resources configures this KafkaConnect
#  # to use KafkaConnector resources to avoid
#  # needing to call the Connect REST API directly
    strimzi.io/use-connector-resources: "true"
spec:
  version: 2.4.0
  replicas: 1
  bootstrapServers: kafka-cluster-kafka-external-bootstrap:9094
  image: ecr-repo/kafkaconnectors:tagname
  config:
    group.id: kafkaconnect-cluster
    offset.storage.topic: kafkaconnect-cluster-offsets
    offset.storage.replication.factor: 1
    config.storage.topic: kafkaconnect-cluster-configs
    config.storage.replication.factor: 1
    status.storage.topic: kafkaconnect-cluster-status
    status.storage.replication.factor: 1
  externalConfiguration:
    env:
      - name: AWS_ACCESS_KEY_ID
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsAccessKey  
      - name: AWS_SECRET_ACCESS_KEY
        valueFrom:
          secretKeyRef:
            name: aws-creds
            key: awsSecretAccessKey
  metrics:
    lowercaseOutputName: true
    lowercaseOutputLabelNames: true
    rules:
    - pattern : "kafka.connect<type=connect-worker-metrics>([^:]+):"
      name: "kafka_connect_connect_worker_metrics_$1"
    - pattern : "kafka.connect<type=connect-metrics, client-id=([^:]+)><>([^:]+)"
      name: "kafka_connect_connect_metrics_$1_$2"

Expected behavior
Would like to be able to use the AvroConverter. I can't be the only one, it must be something I'm doing wrong.

Environment (please complete the following information):

• Strimzi version: 0.17.0
• Installation method: Strimzi operator
• Kubernetes cluster: Kubernetes 1.16
• Infrastructure: Amazon EKS

[scholzj]

TBH, I never used an of these, but one thing is not clear to me ... is the AvroConverterConfig class part of the connector? Or where do you add it to the container image?

[scholzj]

Or maybe just to clarify why I asked that question ... Strimzi is based on Apache Kafka. That is not the same as Confluent Platform. Confluent Platform is based on Apache Kafka, but adds some additional pieces - some are proprietary, some are open source but not part of Apache Kafka and Apache Software Foundation. And while the core messaging is compatible, not everything might transfer easily. So I guess that the AvroConverter classes might be part of Confluent Platform but not of Apache Kafka. So (if the license etc. permits) you will need to add them to the Kafka Connect container image together witht he connector to be able to use them.

[timkalanai]

Right, so both AvroConverter and AvroConverterConfig are maintained by confluent. If i understand the problem correctly, this is some sort of classpath issue (though I can't figure out why). My plugin directory looks like this:

plugins

• kafka-storage-common
  
  
  
  • jars that include both AvroConverter and AvroConverterConfig
  
  
• kafka-connect-s3
  
  
  
  • s3 connector jars
  
  
  • sym link to ../kafka-storage-common
  
  

The symlink is a little weird. So I tried everything from combining all the jars into 1 directory, and even dissecting the jar file, and pasting the class files together into 1 big jar... (went a little crazy)....

Is there possibly some sort of compatibility issue? I tried with confluent 5.1 jars which is supposed to be compatible with kafka 2.1. My strimzi kafka is 2.4.0.

I looked at docker hub, the "strimzi kafka connect" image I'm using is tagged like this: strimzi/kafka-connect:0.11.4-kafka-2.1.0

Does that mean that this image some how is tied to a particular version of kafka?

[timkalanai]

Any chance this has to do with the kafka connect workers running with slightly different class paths? The exception seems to happen when the "Worker" is starting the task

[scholzj]

If you added these classes to the image then yes, the problem can be in
classpath. But they are for sure not part of Apache Kafka ... so for that
to happen you need to add them first.

In general AFAIK all JARs belonging to one connector should be in one
directory or subdirectory.

On Mon, 18 May 2020 at 01:16, timkalanai [e-mail] wrote:

Any chance this has to do with the kafka connect workers running with
slightly different class paths? The exception seems to happen when the
"Worker" is starting the task

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
https://github.com/strimzi/strimzi-kafka-operator/issues/3047#issuecomment-629876341,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/ABLFOR2ZNW54RQDNOMDLM5DRSBV47ANCNFSM4NDSWKYA
.

[timkalanai]

So I got a different error now after I removed the jars with "AvroConverter". Yay! different is good. The following should mean that normally it's finding the jar with AvroConverter:

Tasks:
      Id:     0
      State:  FAILED
      Trace:  org.apache.kafka.common.config.ConfigException: Invalid value io.confluent.connect.avro.AvroConverter for configuration key.converter: Class io.confluent.connect.avro.AvroConverter could not be found.
              at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:718)
              at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:471)
              at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:464)
              at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:62)
              at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:75)
              at org.apache.kafka.connect.runtime.ConnectorConfig.<init>(ConnectorConfig.java:203)
              at org.apache.kafka.connect.runtime.ConnectorConfig.<init>(ConnectorConfig.java:199)
              at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:404)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:873)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1600(DistributedHerder.java:111)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:888)
              at org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:884)
              at java.util.concurrent.FutureTask.run(FutureTask.java:266)
              at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
              at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
              at java.lang.Thread.run(Thread.java:748)

I think it must have something to do with how it's class loading.... There's a little magical switch statement in here that looks suspicious:
https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java#L235

[timkalanai]

I think I figured it out, and it's weird...

I think my problem was with the way that kafka connect scans for "Connectors" versus "Converters". There's a lot of classloading magic in that Plugins file mentioned above. I'm probably not going to do an explanation justice because I don't quite get it myself.

But as it scans through the folder structure (you can see it loading connectors and convertors in the logs as kafka connect starts up), it looks for Converters and Connectors in parallel. That being said, I think because they try to isolate connectors within each directory in the plugins folder, the dependencies for each have to be within each directory (which is what we thought).

Confluent has laid out their folder structure a little differently. There's a commons folder with common libs (I needed common-config.jar and common-config.jar, but there are others in there too). There's also a "kafka-storage-common" that has AvroConverter.

Confluent puts a symlink in the connector directories to make sure that they have access to the right converters. Think when connect is traversing each folder, it does a deep traversal, and a symlink looks like a directory.

I had a number of issues. But I finally solved it by keeping the symlink, and putting the confluent-common jars into the Kafka-storage-common folder. That way, when AvroConverter is detected in the kafka-storage-common folder, it loads in the same class loader as the common jars in the same folder.

And any connector I need "AvroConverter" for, I add a symlink to the "kafka-storage-common" directory.

I know it's convoluted. Maybe I'm just too tired and not seeing straight, but everything seems to be working now. Hope this helps some person in the future.

[scholzj]

Glad that it works for you and thanks a lot for sharing the solution. This would be helpul for anyone else.

I guess we can close this issue?

[timkalanai]

yup! thanks!

[cameronbraid]

I'm trying to do a similar thing, would be nice to see your whole dockerfile :)

[timkalanai]

The key isn't really in the dockerfile which is this:

FROM strimzi/kafka-connect:0.11.4-kafka-2.1.0
USER root:root
COPY ./connect-plugins/ /opt/kafka/plugins/
USER 1001

It's in the way that you structure the directories in your plugin folders (unfortunately) - in my case ./connect-plugins/

Look at 2 things below

• The symbolic link in kafka-connect-s3 which links to ../kafka-connect-storage-common (which is a directory in the confluent platform) this allows the s3 plugin to access common jars that are needed for it to run
• In kafka-connect-storage-common, I had to add all of the confluent "common" jars, which are common-config, common-metrics, common-utils (from confluent-common in the java folder in the confluent platform download)

total 40
drwxr-xr-x    9 someuser  removed    288 May 17 20:16 .
drwxr-xr-x   16 someuser  removed    512 May 17 22:18 ..
drwxr-xr-x@  26 someuser  removed    832 May 17 19:34 kafka-connect-s3
drwxr-xr-x@ 159 someuser  removed   5088 May 17 20:17 kafka-connect-storage-common

./kafka-connect-s3:
total 18296
drwxr-xr-x@ 26 someuser  removed      832 May 17 19:34 .
drwxr-xr-x   9 someuser  removed      288 May 17 20:16 ..
-rw-r--r--@  1 someuser  removed    20437 Apr  2 13:50 audience-annotations-0.5.0.jar
-rw-r--r--@  1 someuser  removed   999670 Apr  2 13:50 aws-java-sdk-core-1.11.725.jar
-rw-r--r--@  1 someuser  removed   592109 Apr  2 13:50 aws-java-sdk-kms-1.11.725.jar
-rw-r--r--@  1 someuser  removed  1047529 Apr  2 13:50 aws-java-sdk-s3-1.11.725.jar
-rw-r--r--@  1 someuser  removed   148009 Apr  2 13:50 aws-java-sdk-sts-1.11.725.jar
-rw-r--r--@  1 someuser  removed    20904 Apr  2 13:50 common-utils-5.1.4.jar
-rw-r--r--@  1 someuser  removed   335042 Apr  2 13:50 commons-codec-1.11.jar
-rw-r--r--@  1 someuser  removed    61829 Apr  2 13:50 commons-logging-1.2.jar
-rw-r--r--@  1 someuser  removed   774384 Apr  2 13:50 httpclient-4.5.9.jar
-rw-r--r--@  1 someuser  removed   326724 Apr  2 13:50 httpcore-4.4.4.jar
-rw-r--r--@  1 someuser  removed   565410 Apr  2 13:50 ion-java-1.0.2.jar
-rw-r--r--@  1 someuser  removed    66894 Apr  2 13:50 jackson-annotations-2.9.10.jar
-rw-r--r--@  1 someuser  removed   325636 Apr  2 13:50 jackson-core-2.9.10.jar
-rw-r--r--@  1 someuser  removed  1348786 Apr  2 13:50 jackson-databind-2.9.10.3.jar
-rw-r--r--@  1 someuser  removed    50986 Apr  2 13:50 jackson-dataformat-cbor-2.9.10.jar
-rw-r--r--@  1 someuser  removed   213911 Apr  2 13:50 jline-2.12.1.jar
-rw-r--r--@  1 someuser  removed    27586 Apr  2 13:50 jmespath-java-1.11.725.jar
-rw-r--r--@  1 someuser  removed    73631 Apr  2 13:50 kafka-connect-s3-5.1.4.jar
-rw-r--r--@  1 someuser  removed  1292696 Apr  2 13:50 netty-3.10.6.Final.jar
-rw-r--r--@  1 someuser  removed    41203 Apr  2 13:50 slf4j-api-1.7.25.jar
lrwxr-xr-x   1 someuser  removed       31 Apr  2 13:50 storage-common -> ../kafka-connect-storage-common
-rw-r--r--@  1 someuser  removed    74798 Apr  2 13:50 zkclient-0.10.jar
-rw-r--r--@  1 someuser  removed   906708 Apr  2 13:50 zookeeper-3.4.13.jar

./kafka-connect-storage-common:
total 233536
drwxr-xr-x@ 159 someuser  removed      5088 May 17 20:17 .
drwxr-xr-x    9 someuser  removed       288 May 17 20:16 ..
-rw-r--r--@   1 someuser  removed     23698 Apr  2 13:09 common-config-5.1.4.jar
-rw-r--r--@   1 someuser  removed     41450 Apr  2 13:09 common-metrics-5.1.4.jar
-rw-r--r--@   1 someuser  removed     20904 Apr  2 13:09 common-utils-5.1.4.jar
-rw-r--r--@   1 someuser  removed    188671 Apr  2 13:40 commons-beanutils-1.7.0.jar
.......
-rw-r--r--@   1 someuser  removed     34997 Apr  2 13:40 kafka-avro-serializer-5.1.4.jar
-rw-r--r--@   1 someuser  removed   1899410 Apr  2 13:40 kafka-clients-2.1.1-cp6.jar
-rw-r--r--@   1 someuser  removed     46863 Apr  2 13:40 kafka-connect-avro-converter-5.1.4.jar
-rw-r--r--@   1 someuser  removed      7627 Apr  2 13:40 kafka-connect-storage-common-5.1.4.jar
-rw-r--r--@   1 someuser  removed     26196 Apr  2 13:40 kafka-connect-storage-core-5.1.4.jar
-rw-r--r--@   1 someuser  removed      4371 Apr  2 13:40 kafka-connect-storage-format-5.1.4.jar
-rw-r--r--@   1 someuser  removed     32971 Apr  2 13:40 kafka-connect-storage-hive-5.1.4.jar
-rw-r--r--@   1 someuser  removed     26376 Apr  2 13:40 kafka-connect-storage-partitioner-5.1.4.jar
-rw-r--r--@   1 someuser  removed      2955 Apr  2 13:40 kafka-connect-storage-wal-5.1.4.jar
-rw-r--r--@   1 someuser  removed     54234 Apr  2 13:40 kafka-schema-registry-client-5.1.4.jar
-rw-r--r--@   1 someuser  removed    442805 Apr  2 13:40 kryo-2.22.jar
-rw-r--r--@   1 someuser  removed   1045744 Apr  2 13:40 leveldbjni-all-1.8.jar
.....

It's not a great solution, and I tried to explain why it works above. But it's what's worked for me so far.

[cameronbraid]

@timkalanai thanks mate - thats greatly appreciated, with your help I was able to get it working - thanks again

here's the Dockerfile i use

FROM confluentinc/cp-kafka-connect:5.5.0 as cp
FROM strimzi/kafka:0.18.0-kafka-2.5.0
USER root:root
COPY --from=cp /usr/share/java/kafka-connect-storage-common /opt/kafka/plugins/kafka-connect-storage-common
COPY --from=cp /usr/share/java/confluent-common /opt/kafka/plugins/confluent-common
COPY --from=cp /usr/share/java/kafka-connect-s3 /opt/kafka/plugins/kafka-connect-s3
COPY --from=cp /usr/share/java/kafka-connect-jdbc /opt/kafka/plugins/kafka-connect-jdbc
RUN cd /opt/kafka/plugins && for plugin in kafka-connect-s3 kafka-connect-jdbc; do cd $plugin; ln -s ../confluent-common; ln -s ../kafka-connect-storage-common; cd ..; done