Strimzi-kafka-operator: How to init kafka connect with config

[jslusher]

I have a KafkaConnect deployment with a docker image that uses the strimzi image as its source and then installs the debezium mysql plugin. I have the following in the config: section:

  config:
    connector.class: io.debezium.connector.mysql.MySqlConnector
    tasks.max: 1
    database.hostname: debezium-mysql.svc-mysql
    database.port: 3306
    database.user: debezium
    database.password: "${DEBEZIUM_PASSWORD}"
    database.server.id: 1701
    database.server.name: mysql
    database.whitelist: mysql
    database.history.kafka.bootstrap.servers: debezium-kafka-bootstrap:9092
    database.history.kafka.topic: my-topic
    table.whitelist: mysql.my_table

The connect pod will come up, but I have to run a separate POST command afterwards with the above configuration to initiate the plugin and start the debezium snapshot process. Is there something I'm doing wrong there or is this by design?

[scholzj]

This is by design. Kafka Connect can work in two modes - the so called standalone and distributed. The Kafka Connect deployment from Strimzi is using the distributed mode in which the connectors are configured through the REST interface. The reason why we use the distributed mode is that the standalone mode is using disk to store the offsets etc. and that makes is a bit harder to handle on Kubernetes.

[jslusher]

So does that mean the above config options aren't doing anything? What would you recommend as the most integrated way to have the config applied through the REST interface? Is there a way to that using operator features or should I come up with something that would sit alongside the operator configuration?

[scholzj]

We have plans for _operator for connectors_ which will do that. But it doesn't exist today. So for now it is a bit crude - you can for example just use curl. Here is an example how to do it with _one command_ (you would need to replace it with your plugin and your configuration and you might need to change the label selector in the command).

kubectl exec -ti $(kubectl get pod -l app=my-connect-cluster -o=jsonpath='{.items[0].metadata.name}') -- curl -X POST -H "Content-Type: application/json" --data '{ "name": "sink-test", "config": { "connector.class": "FileStreamSink", "tasks.max": "3", "topics": "my-topic", "file": "/tmp/test.sink.txt" } }' http://localhost:8083/connectors

[jslusher]

I made a little script that will take care of it in the event we need it. Thank you for the responses!

[dhessler]

@scholzj could this be supported by updating https://github.com/strimzi/strimzi-kafka-operator/blob/0b76ba0b3c7926a840c7e1f51c8473f4c9567ecd/docker-images/kafka/scripts/kafka_connect_config_generator.sh#L85
to

$(echo -e ${KAFKA_CONNECT_CONFIGURATION})

so that newlines in the config are properly formatted for the properties files?

The current implementation is broken for configuring more than a single property, e.g.:
export KAFKA_CONNECT_CONFIGURATION=connector.class=io.debezium.connector.mysql.MySqlConnector\ntasks.max=1\ndatabase.hostname=debezium-mysql.svc-mysql

Same goes for the variable expansion of KAFKA_MIRRORMAKER_CONFIGURATION_PRODUCER, KAFKA_MIRRORMAKER_CONFIGURATION_CONSUMER, KAFKA_CONNECT_CONFIGURATION, KAFKA_CONFIGURATION, ZOOKEEPER_CONFIGURATION in docker-images/kafka/scripts.

[scholzj]

@dhessler I'm not sure I understand the problem. You do not configure the KAFKA_CONNECT_CONFIGURATION environment variable by exporting it. It is configured through the KafkaConnect.spec.config (see docs). And it works fine with multiple options specified.

[dhessler]

In my case I was trying to create a custom docker image (FROM strimzi/kafka-connect:0.11.1-kafka-2.1.0), but noticed that the KAFKA_CONNECT_CONFIGURATION env var does not work with newlines (i.e., multiple options).

With the above update to kafka_connect_config_generator.sh, kafka connect ran with the expected config.

Likewise, when deploying to kubernetes, I noticed the config was not being picked up through KafkaConnect.spec.config yaml. The curl command you mentioned seems to work, but I'd like to use the yaml config for ease of deployment. Am I doing something wrong? Maybe I need to update to the latest kafka-connect:0.11.4-kafka-2.1.0?

[scholzj]

Likewise, when deploying to kubernetes, I noticed the config was not being picked up through KafkaConnect.spec.config yaml. The curl command you mentioned seems to work, but I'd like to use the yaml config for ease of deployment. Am I doing something wrong? Maybe I need to update to the latest kafka-connect:0.11.4-kafka-2.1.0?

This should work and I using it from time to time. If this doesn't work, it might be a bug. Could you please provide an example of the KafkaConnect custom resource where it didn't worked for you? I will have a look at it.

Maybe I need to update to the latest kafka-connect:0.11.4-kafka-2.1.0?

I don't think this matters. There were only little differences between 0.11.1 and 0.11.4. I do not think any of the bug-fixes were related to this.

[dhessler]

Thank you @scholzj, I've tested again (on 0.11.1) and the config indeed appears to work. With the exception however of the "partitioner.class" property, which fails with the following error when set:

unbundler-worker-connect-668f8c5dbd-wdvtx unbundler-worker-connect 2019-06-11 07:31:23,410 ERROR Uncaught exception in herder work thread, exiting:  (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder]                                                 
unbundler-worker-connect-668f8c5dbd-wdvtx unbundler-worker-connect org.apache.kafka.common.config.ConfigException: Invalid value io.confluent.connect.storage.partitioner.TimeBasedPartitioner for configuration partitioner.class: Class io.confluent.connect.storage.partitioner.TimeBas
edPartitioner could not be found.

Here is my spec:

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaConnect
metadata:
  name: {{ .Chart.Name }}
spec:
  version: 2.2.1
  replicas: {{ .Values.kafkaS3Connect.replicas | int }}
  image: "{{ .Values.kafkaS3Connect.repository }}:{{ .Values.global.tag }}"
  bootstrapServers: {{ .Values.kafka.broker }}-{{ .Release.Namespace }}-kafka-bootstrap:{{ .Values.kafka.bootstrapPort }}
  config:
    group.id: {{ .Chart.Name }}
    connector.class: io.confluent.connect.s3.S3SinkConnector
    config.storage.topic: {{ .Chart.Name }}-configs
    offset.storage.topic: {{ .Chart.Name }}-offsets
    status.storage.topic: {{ .Chart.Name }}-statuses
    flush.size: 1000
    format.bytearray.extension: .json
    format.class: io.confluent.connect.s3.format.json.JsonFormat
    key.converter: org.apache.kafka.connect.storage.StringConverter
    key.converter.schemas.enable: false
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: false
    locale: en
    partition.duration.ms: 3600000
    partitioner.class: io.confluent.connect.storage.partitioner.TimeBasedPartitioner
    path.format: YYYY/MM/dd/HH/mm
    s3.bucket.name: {{ .Values.configVar.S3_BUCKET }}
    s3.region: {{ .Values.configVar.S3_REGION }}
    schema.compatibility: NONE
    storage.class: io.confluent.connect.s3.storage.S3Storage
    tasks.max: 1
    timestamp.extractor: RecordField
    timestamp.field: ingestionTimestamp
    timezone: UTC
    topics: {{ .Values.configVar.KAFKA_TOPIC_DATA_POINTS }}

The image was built through this dockerfile:
```
FROM strimzi/kafka-connect:0.11.1-kafka-2.1.0
USER root:root

RUN yum -y update && yum -y install wget

RUN wget -q https://archive.landoop.com/lkd/packages/connectors/third-party/kafka-connect-s3/kafka-connect-s3-5.2.1-lkd-r0.tar.gz -O /opt/kafka-connect-s3.tar.gz \
&& mkdir -p /opt/kafka/plugins/ \
&& tar --no-same-owner -xf /opt/kafka-connect-s3.tar.gz -C /opt/kafka/plugins/ \
&& rm -rf /opt/kafka-connect-s3.tar.gz

USER 1001
````

Kafka connect runs fine with the above config when I omit the partitioner.class property (the other io.confluent.connect.* classes are found). Oddly, when I set the TimeBasedPartitioner via the curl/kubectl command you provided it works! Is there something that the API is doing that the config spec doesn't support? Do you see anything wrong with my setup?

Sorry if this is heading OT, I can open a new issue if it helps.

[scholzj]

I have very little experience with this particular connector. But the partitioner.class as you used it will be IMHO taken by the Kafka Producer used to send data from Kafka Connect to the Kafka broker when Source plugins are used. And I think that the TimeBasedPartitioner is not that kind of partitioner but some partitioner which defines how will the data be stored in S3.

In general, Strimzi runs Kafka in distributed mode and I do not think you should configure the connector in the spec.config field. You should use only the generic Kafka Connect configuration there. The plugin configuration has to be done over the REST interface - that is why when you use it there, it understands what does the partitioner.class mean and doesn't complain about it.

[dhessler]

Aha, I see. Thank you for looking into this.

I had been trying to debug this issue at length, to no avail. The errors were not very helpful (here for reference):

[2019-06-14 18:32:06,604] DEBUG [Thread[org.reflections-scanner-0,5,main]] scanning file:/opt/kafka/plugins/confluentinc-kafka-connect-s3-5.0.0/lib/kafka-connect-storage-partitioner-5.0.0.jar (Reflections:197)
[2019-06-14 18:32:06,605] DEBUG could not scan file META-INF/MANIFEST.MF in url file:/opt/kafka/plugins/confluentinc-kafka-connect-s3-5.0.0/lib/kafka-connect-storage-partitioner-5.0.0.jar with scanner SubTypesScanner (Reflections:257)
org.reflections.ReflectionsException: could not create class object from file META-INF/MANIFEST.MF
  at org.reflections.scanners.AbstractScanner.scan(AbstractScanner.java:32)
  at org.reflections.Reflections.scan(Reflections.java:253)
  at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader$InternalReflections.scan(DelegatingClassLoader.java:412)
  at org.reflections.Reflections$1.run(Reflections.java:198)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  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)
2019-06-14T18:32:06.60707385Z Caused by: org.reflections.ReflectionsException: could not create class file from MANIFEST.MF
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:102)
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:24)
  at org.reflections.scanners.AbstractScanner.scan(AbstractScanner.java:30)
  ... 8 more
Caused by: java.io.IOException: bad magic number: 4d616e69
  at javassist.bytecode.ClassFile.read(ClassFile.java:835)
  at javassist.bytecode.ClassFile.<init>(ClassFile.java:164)
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:100)
  ... 10 more
2019-06-14T18:32:06.60716112Z [2019-06-14 18:32:06,605] DEBUG could not scan file META-INF/maven/io.confluent/kafka-connect-storage-partitioner/pom.xml in url file:/opt/kafka/plugins/confluentinc-kafka-connect-s3-5.0.0/lib/kafka-connect-storage-partitioner-5.0.0.jar with scanner SubTypesScanner (Reflections:257)
org.reflections.ReflectionsException: could not create class object from file META-INF/maven/io.confluent/kafka-connect-storage-partitioner/pom.xml
  at org.reflections.scanners.AbstractScanner.scan(AbstractScanner.java:32)
  at org.reflections.Reflections.scan(Reflections.java:253)
  at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader$InternalReflections.scan(DelegatingClassLoader.java:412)
  at org.reflections.Reflections$1.run(Reflections.java:198)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  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)
Caused by: org.reflections.ReflectionsException: could not create class file from pom.xml
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:102)
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:24)
  at org.reflections.scanners.AbstractScanner.scan(AbstractScanner.java:30)
  ... 8 more
Caused by: java.io.IOException: bad magic number: 3c212d2d
  at javassist.bytecode.ClassFile.read(ClassFile.java:835)
  at javassist.bytecode.ClassFile.<init>(ClassFile.java:164)
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:100)
  ... 10 more
2019-06-14T18:32:06.6072687Z [2019-06-14 18:32:06,605] DEBUG could not scan file META-INF/maven/io.confluent/kafka-connect-storage-partitioner/pom.properties in url file:/opt/kafka/plugins/confluentinc-kafka-connect-s3-5.0.0/lib/kafka-connect-storage-partitioner-5.0.0.jar with scanner SubTypesScanner (Reflections:257)
org.reflections.ReflectionsException: could not create class object from file META-INF/maven/io.confluent/kafka-connect-storage-partitioner/pom.properties
  at org.reflections.scanners.AbstractScanner.scan(AbstractScanner.java:32)
  at org.reflections.Reflections.scan(Reflections.java:253)
  at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader$InternalReflections.scan(DelegatingClassLoader.java:412)
  at org.reflections.Reflections$1.run(Reflections.java:198)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  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)
Caused by: org.reflections.ReflectionsException: could not create class file from pom.properties
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:102)
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:24)
  at org.reflections.scanners.AbstractScanner.scan(AbstractScanner.java:30)
  ... 8 more
Caused by: java.io.IOException: bad magic number: 23437265
  at javassist.bytecode.ClassFile.read(ClassFile.java:835)
  at javassist.bytecode.ClassFile.<init>(ClassFile.java:164)
  at org.reflections.adapters.JavassistAdapter.getOfCreateClassObject(JavassistAdapter.java:100)
  ... 10 more

Seems like https://github.com/strimzi/strimzi-kafka-operator/issues/1468 would add support for configuring individual connectors.
It wasn't clear to me that the spec.config only supports generic kafka connector configuration. Consider improving the kafka connect plugins docs regarding configuration, with a warning to use the connect rest api directly for connector-specific configuration properties.

Thanks again!

[scholzj]

@dhessler Thanks. I will add your comments about the docs to our backlog and will try to look at it.