Confluent Datagen Source Kafka Connector Deployment Error

[kotapullarao]

Hello Community,

I am trying to deploy a Datagen Source connector from Confluent in Strimzi Kafka connect which is running on OCP k8s. Our Kafka cluster resides in Confluent Cloud. I am building a Custom Strimzi Kafka connect image which includes necessary connectors/plugins.

In Kafka connect instance when I describe as below it lists the available connectors/plugins which contains Datagen as well but when I deploy the connector it fails to find any such connector class.

Below is the main error log and Docker file and CRD's. I have attached the full log as well. Kindly help me if you have any inputs to solve this issue.

Custom Docker file for Image

FROM confluentinc/cp-kafka-connect:latest AS cp

RUN for i in datagen avro-converter elasticsearch; do \
  confluent-hub install --no-prompt confluentinc/kafka-connect-"$i":latest & \
  done \
  && wait

FROM quay.io/strimzi/kafka:latest-kafka-3.8.0

USER root:root


RUN for i in datagen avro-converter elasticsearch; do \
  mkdir -p /opt/kafka/plugins/kafka-connect-"$i" & \
  done \
  && wait


ARG PLUGIN_SRC=/usr/share/confluent-hub-components/confluentinc-kafka-connect-
ARG PLUGIN_DEST=/opt/kafka/plugins/kafka-connect-


COPY --from=cp ${PLUGIN_SRC}datagen/ ${PLUGIN_DEST}datagen
COPY --from=cp ${PLUGIN_SRC}avro-converter/ ${PLUGIN_DEST}avro-converter
COPY --from=cp ${PLUGIN_SRC}elasticsearch/ ${PLUGIN_DEST}elasticsearch


ENV CLASSPATH="/opt/kafka/plugins/*"


USER 1001

Kafka Connect crd yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  annotations:
    argocd.argoproj.io/tracking-id: 'dev2.project.strimzi-kafka-connect:kafka.strimzi.io/KafkaConnect:project/strimzi-kafka-connect'
    kubectl.kubernetes.io/last-applied-configuration: |
      {"apiVersion":"kafka.strimzi.io/v1beta2","kind":"KafkaConnect","metadata":{"annotations":{"argocd.argoproj.io/tracking-id":"dev2.project.strimzi-kafka-connect:kafka.strimzi.io/KafkaConnect:project/strimzi-kafka-connect","strimzi.io/use-connector-resources":"true"},"labels":{"app.kubernetes.io/instance":"project-strimzi-kafka-connect","app.kubernetes.io/managed-by":"Helm","app.kubernetes.io/name":"strimzi-kafka-connect","app.kubernetes.io/version":"strimzi-connector_34-SNAPSHOT","helm.sh/chart":"strimzi-connect-0.1.0-strimzi-662"},"name":"strimzi-kafka-connect","namespace":"project"},"spec":{"authentication":{"passwordSecret":{"password":"password","secretName":"strimzi-cloud-secret"},"type":"plain","username":"USERNAME"},"bootstrapServers":"pksf-1fgddgfgvj.westeurope.azure.confluent.cloud:9092","config":{"config.providers":"file","config.providers.file.class":"org.apache.kafka.common.config.provider.FileConfigProvider","plugin.discovery":"only_scan"},"externalConfiguration":{"volumes":[{"name":"confluent-elastic-secret","secret":{"secretName":"confluent-elastic-secret"}},{"name":"confluent-keyvault-secret","secret":{"secretName":"confluent-keyvault-secret"}}]},"image":"ghcr.io/dvdvfgg/strimzi-kafka-connect:strimzi-connector_34-SNAPSHOT","replicas":1,"tls":{"trustedCertificates":[]}}}
    strimzi.io/use-connector-resources: 'true'
  resourceVersion: '572734311'
  name: strimzi-kafka-connect
  uid: 635350a3-d02d-4e1c-83de-ab43accbd523
  creationTimestamp: '2024-10-17T09:25:52Z'
  generation: 2
  managedFields:
    - apiVersion: kafka.strimzi.io/v1beta2
      fieldsType: FieldsV1
      fieldsV1:
        'f:metadata':
          'f:annotations':
            .: {}
            'f:argocd.argoproj.io/tracking-id': {}
            'f:kubectl.kubernetes.io/last-applied-configuration': {}
            'f:strimzi.io/use-connector-resources': {}
          'f:labels':
            .: {}
            'f:app.kubernetes.io/instance': {}
            'f:app.kubernetes.io/managed-by': {}
            'f:app.kubernetes.io/name': {}
            'f:app.kubernetes.io/version': {}
            'f:helm.sh/chart': {}
        'f:spec':
          .: {}
          'f:authentication':
            .: {}
            'f:passwordSecret':
              .: {}
              'f:password': {}
              'f:secretName': {}
            'f:type': {}
            'f:username': {}
          'f:bootstrapServers': {}
          'f:config':
            .: {}
            'f:plugin.discovery': {}
          'f:externalConfiguration':
            .: {}
            'f:volumes': {}
          'f:image': {}
          'f:replicas': {}
          'f:tls':
            .: {}
            'f:trustedCertificates': {}
      manager: argocd-controller
      operation: Update
      time: '2024-10-17T09:25:52Z'
    - apiVersion: kafka.strimzi.io/v1beta2
      fieldsType: FieldsV1
      fieldsV1:
        'f:status':
          .: {}
          'f:conditions': {}
          'f:connectorPlugins': {}
          'f:labelSelector': {}
          'f:observedGeneration': {}
          'f:replicas': {}
          'f:url': {}
      manager: strimzi-cluster-operator
      operation: Update
      subresource: status
      time: '2024-10-17T09:39:57Z'
  namespace: project
  labels:
    app.kubernetes.io/instance: project-strimzi-kafka-connect
    app.kubernetes.io/managed-by: Helm
    app.kubernetes.io/name: strimzi-kafka-connect
    app.kubernetes.io/version: strimzi-connector_34-SNAPSHOT
    helm.sh/chart: strimzi-connect-0.1.0-strimzi-662
spec:
  authentication:
    passwordSecret:
      password: password
      secretName: strimzi-cloud-secret
    type: plain
    username: USERNAME
  bootstrapServers: 'pksf-1fgddgfgvj.westeurope.azure.confluent.cloud:9092'
  config:
    plugin.discovery: only_scan
  externalConfiguration:
    volumes:
      - name: confluent-elastic-secret
        secret:
          secretName: confluent-elastic-secret
      - name: confluent-keyvault-secret
        secret:
          secretName: confluent-keyvault-secret
  image: 'ghcr.io/dvdvfgg/strimzi-kafka-connect:strimzi-connector_34-SNAPSHOT'
  replicas: 1
  tls:
    trustedCertificates: []
status:
  conditions:
    - lastTransitionTime: '2024-10-17T09:39:57.361679366Z'
      status: 'True'
      type: Ready
  connectorPlugins:
    - class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
      type: sink
      version: 14.1.1
    - class: io.confluent.kafka.connect.datagen.DatagenConnector
      type: source
      version: 'null'
    - class: org.apache.kafka.connect.mirror.MirrorCheckpointConnector
      type: source
      version: 3.8.0
    - class: org.apache.kafka.connect.mirror.MirrorHeartbeatConnector
      type: source
      version: 3.8.0
    - class: org.apache.kafka.connect.mirror.MirrorSourceConnector
      type: source
      version: 3.8.0
  labelSelector: 'strimzi.io/cluster=strimzi-kafka-connect,strimzi.io/name=strimzi-kafka-connect-connect,strimzi.io/kind=KafkaConnect'
  observedGeneration: 2
  replicas: 1
  url: 'http://strimzi-kafka-connect-connect-api.project.svc:8083'

Kafka Connector crd yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  annotations:
    argocd.argoproj.io/tracking-id: 'dev2.project.confluent-connectors:kafka.strimzi.io/KafkaConnector:project/datagen-source-connector'
    kubectl.kubernetes.io/last-applied-configuration: |
      {"apiVersion":"kafka.strimzi.io/v1beta2","kind":"KafkaConnector","metadata":{"annotations":{"argocd.argoproj.io/tracking-id":"dev2.project.confluent-connectors:kafka.strimzi.io/KafkaConnector:project/datagen-source-connector"},"labels":{"app.kubernetes.io/instance":"project-confluent-connectors","app.kubernetes.io/managed-by":"Helm","app.kubernetes.io/name":"datagen-source-connector","helm.sh/chart":"datagen-source-connector-0.1.0-strimzi-657","strimzi.io/cluster":"strimzi-kafka-connect"},"name":"datagen-source-connector","namespace":"project"},"spec":{"class":"io.confluent.kafka.connect.datagen.DatagenConnector","config":{"connector.class":"io.confluent.kafka.connect.datagen.DatagenConnector","iterations":"10000000","kafka.api.key":"${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:kafka-cluster-api-key}","kafka.api.secret":"${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:kafka-cluster-api-secret}","kafka.auth.mode":"KAFKA_API_KEY","kafka.topic":"inventory","key.converter":"io.confluent.connect.avro.AvroConverter","key.converter.basic.auth.credentials.source":"USER_INFO","key.converter.basic.auth.user.info":"${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}","key.converter.schema.enable":"true","key.converter.schema.registry.basic.auth.user.info":"${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}","key.converter.schema.registry.url":"https://psffffc-dsdgsv.westeurope.azure.confluent.cloud","max.interval":"1000","name":"datagen-source-connector","quickstart":"inventory","schema.registry.url":"https://psffffc-dsdgsv.westeurope.azure.confluent.cloud","topics":"inventory","type.name":"kafka-connect","value.converter":"io.confluent.connect.avro.AvroConverter","value.converter.basic.auth.credentials.source":"USER_INFO","value.converter.basic.auth.user.info":"${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}","value.converter.schema.registry.basic.auth.user.info":"${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}","value.converter.schema.registry.url":"https://psffffc-dsdgsv.westeurope.azure.confluent.cloud"},"tasksMax":1}}
  resourceVersion: '572794046'
  name: datagen-source-connector
  uid: d8f2e1a2-ae01-4d75-a93f-73f6e7819ebe
  creationTimestamp: '2024-10-17T11:14:01Z'
  generation: 1
  managedFields:
    - apiVersion: kafka.strimzi.io/v1beta2
      fieldsType: FieldsV1
      fieldsV1:
        'f:metadata':
          'f:annotations':
            .: {}
            'f:argocd.argoproj.io/tracking-id': {}
            'f:kubectl.kubernetes.io/last-applied-configuration': {}
          'f:labels':
            .: {}
            'f:app.kubernetes.io/instance': {}
            'f:app.kubernetes.io/managed-by': {}
            'f:app.kubernetes.io/name': {}
            'f:helm.sh/chart': {}
            'f:strimzi.io/cluster': {}
        'f:spec':
          .: {}
          'f:class': {}
          'f:config':
            'f:kafka.api.secret': {}
            'f:key.converter.schema.registry.basic.auth.user.info': {}
            'f:key.converter.basic.auth.credentials.source': {}
            'f:max.interval': {}
            'f:connector.class': {}
            'f:quickstart': {}
            'f:value.converter.basic.auth.user.info': {}
            'f:iterations': {}
            'f:type.name': {}
            'f:value.converter.schema.registry.url': {}
            'f:kafka.auth.mode': {}
            .: {}
            'f:value.converter.schema.registry.basic.auth.user.info': {}
            'f:kafka.api.key': {}
            'f:value.converter': {}
            'f:kafka.topic': {}
            'f:value.converter.basic.auth.credentials.source': {}
            'f:schema.registry.url': {}
            'f:topics': {}
            'f:key.converter.schema.registry.url': {}
            'f:key.converter.basic.auth.user.info': {}
            'f:key.converter': {}
            'f:name': {}
            'f:key.converter.schema.enable': {}
          'f:tasksMax': {}
      manager: argocd-controller
      operation: Update
      time: '2024-10-17T11:14:01Z'
    - apiVersion: kafka.strimzi.io/v1beta2
      fieldsType: FieldsV1
      fieldsV1:
        'f:status':
          .: {}
          'f:conditions': {}
          'f:observedGeneration': {}
          'f:tasksMax': {}
      manager: strimzi-cluster-operator
      operation: Update
      subresource: status
      time: '2024-10-17T11:14:02Z'
  namespace: project
  labels:
    app.kubernetes.io/instance: project-confluent-connectors
    app.kubernetes.io/managed-by: Helm
    app.kubernetes.io/name: datagen-source-connector
    helm.sh/chart: datagen-source-connector-0.1.0-strimzi-657
    strimzi.io/cluster: strimzi-kafka-connect
spec:
  class: io.confluent.kafka.connect.datagen.DatagenConnector
  config:
    value.converter.schema.registry.url: 'https://psffffc-dsdgsv.westeurope.azure.confluent.cloud'
    kafka.auth.mode: KAFKA_API_KEY
    value.converter: io.confluent.connect.avro.AvroConverter
    kafka.api.key: '${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:kafka-cluster-api-key}'
    topics: inventory
    kafka.topic: inventory
    value.converter.basic.auth.credentials.source: USER_INFO
    key.converter.basic.auth.user.info: '${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}'
    name: datagen-source-connector
    schema.registry.url: 'https://psffffc-dsdgsv.westeurope.azure.confluent.cloud'
    kafka.api.secret: '${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:kafka-cluster-api-secret}'
    key.converter.schema.registry.basic.auth.user.info: '${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}'
    key.converter.schema.registry.url: 'https://psffffc-dsdgsv.westeurope.azure.confluent.cloud'
    max.interval: '1000'
    key.converter: io.confluent.connect.avro.AvroConverter
    key.converter.schema.enable: 'true'
    quickstart: inventory
    iterations: '10000000'
    value.converter.basic.auth.user.info: '${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}'
    key.converter.basic.auth.credentials.source: USER_INFO
    connector.class: io.confluent.kafka.connect.datagen.DatagenConnector
    value.converter.schema.registry.basic.auth.user.info: '${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}'
    type.name: kafka-connect
  tasksMax: 1
status:
  conditions:
    - lastTransitionTime: '2024-10-17T11:14:02.044010008Z'
      message: 'PUT /connectors/datagen-source-connector/config returned 500 (Internal Server Error): Failed to find any class that implements Connector and which name matches io.confluent.kafka.connect.datagen.DatagenConnector, available connectors are: PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorCheckpointConnector, name=''org.apache.kafka.connect.mirror.MirrorCheckpointConnector'', version=''3.8.0'', encodedVersion=3.8.0, type=source, typeName=''source'', location=''classpath''}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorHeartbeatConnector, name=''org.apache.kafka.connect.mirror.MirrorHeartbeatConnector'', version=''3.8.0'', encodedVersion=3.8.0, type=source, typeName=''source'', location=''classpath''}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorSourceConnector, name=''org.apache.kafka.connect.mirror.MirrorSourceConnector'', version=''3.8.0'', encodedVersion=3.8.0, type=source, typeName=''source'', location=''classpath''}'
      reason: ConnectRestException
      status: 'True'
      type: NotReady
  observedGeneration: 1
  tasksMax: 1

Curl response for available connectors/plugins

bash-5.1$ curl http://localhost:8083/connector-plugins
[{"class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","type":"sink","version":"14.1.1"},{"class":"io.confluent.kafka.connect.datagen.DatagenConnector","type":"source","version":"null"},{"class":"org.apache.kafka.connect.mirror.MirrorCheckpointConnector","type":"source","version":"3.8.0"},{"class":"org.apache.kafka.connect.mirror.MirrorHeartbeatConnector","type":"source","version":"3.8.0"},{"class":"org.apache.kafka.connect.mirror.MirrorSourceConnector","type":"source","version":"3.8.0"}]bash-5.1$

Error - For Full Error Log Please see at the end below

2024-10-17 09:40:32,721 INFO AbstractConfig values: 
 (org.apache.kafka.common.config.AbstractConfig) [pool-2-thread-1]
2024-10-17 09:40:32,848 ERROR Error forwarding REST request (org.apache.kafka.connect.runtime.rest.RestClient) [qtp572980523-30]
org.apache.kafka.connect.runtime.rest.errors.ConnectRestException: Failed to find any class that implements Connector and which name matches io.confluent.kafka.connect.datagen.DatagenConnector, available connectors are: PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorCheckpointConnector, name='org.apache.kafka.connect.mirror.MirrorCheckpointConnector', version='3.8.0', encodedVersion=3.8.0, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorHeartbeatConnector, name='org.apache.kafka.connect.mirror.MirrorHeartbeatConnector', version='3.8.0', encodedVersion=3.8.0, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorSourceConnector, name='org.apache.kafka.connect.mirror.MirrorSourceConnector', version='3.8.0', encodedVersion=3.8.0, type=source, typeName='source', location='classpath'}

Connector Config

datagen-source-connector:
  spec:
    class: io.confluent.kafka.connect.datagen.DatagenConnector
    config:
      name: datagen-source-connector
      connector.class: io.confluent.kafka.connect.datagen.DatagenConnector
      topics: inventory
      kafka.topic: "inventory"
      quickstart: "inventory"
      max.interval: "1000"
      iterations: "10000000"
      tasks.max: "1"
      key.converter: io.confluent.connect.avro.AvroConverter
      value.converter: io.confluent.connect.avro.AvroConverter
      key.converter.schema.enable: true
      type.name: kafka-connect
      kafka.auth.mode: KAFKA_API_KEY
      kafka.api.key: ${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:kafka-cluster-api-key}
      kafka.api.secret: ${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:kafka-cluster-api-secret}
      key.converter.basic.auth.credentials.source: "USER_INFO"
      value.converter.basic.auth.credentials.source: "USER_INFO"
      key.converter.schema.registry.basic.auth.user.info: ${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}
      value.converter.schema.registry.basic.auth.user.info: ${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}
      key.converter.basic.auth.user.info: ${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}
      value.converter.basic.auth.user.info: ${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-key}:${file:/opt/kafka/external-configuration/confluent-keyvault-secret/secret.properties:schema-registry-api-secret}

Complete Error Log

Starting Kafka Connect with configuration:
# Bootstrap servers
bootstrap.servers=pksf-1fgddgfgvj.westeurope.azure.confluent.cloud:9092
# REST Listeners
rest.port=8083
rest.advertised.host.name=strimzi-kafka-connect-connect-0.strimzi-kafka-connect-connect.confluent-cicd.svc
rest.advertised.port=8083
# Plugins
plugin.path=/opt/kafka/plugins
# Provided configuration
offset.storage.topic=connect-cluster-offsets
value.converter=org.apache.kafka.connect.json.JsonConverter
config.storage.topic=connect-cluster-configs
key.converter=org.apache.kafka.connect.json.JsonConverter
group.id=connect-cluster
status.storage.topic=connect-cluster-status
plugin.discovery=only_scan


security.protocol=SASL_SSL
producer.security.protocol=SASL_SSL
consumer.security.protocol=SASL_SSL
admin.security.protocol=SASL_SSL


sasl.mechanism=PLAIN
sasl.jaas.config=[hidden]

producer.sasl.mechanism=PLAIN
producer.sasl.jaas.config=[hidden]

consumer.sasl.mechanism=PLAIN
consumer.sasl.jaas.config=[hidden]

admin.sasl.mechanism=PLAIN
admin.sasl.jaas.config=[hidden]

# Additional configuration
consumer.client.rack=

+ exec /usr/bin/tini -w -e 143 -- /opt/kafka/bin/connect-distributed.sh /tmp/strimzi-connect.properties
2024-10-17 09:38:52,072 INFO Kafka Connect worker initializing ... (org.apache.kafka.connect.cli.AbstractConnectCli) [main]
2024-10-17 09:38:52,077 INFO WorkerInfo values: 
	jvm.args = -Xms128M, -XX:+UseG1GC, -XX:MaxGCPauseMillis=20, -XX:InitiatingHeapOccupancyPercent=35, -XX:+ExplicitGCInvokesConcurrent, -XX:MaxInlineLevel=15, -Djava.awt.headless=true, -Dcom.sun.management.jmxremote=true, -Dcom.sun.management.jmxremote.authenticate=false, -Dcom.sun.management.jmxremote.ssl=false, -Dkafka.logs.dir=/opt/kafka, -Dlog4j.configuration=file:/opt/kafka/custom-config/log4j.properties
	jvm.spec = Red Hat, Inc., OpenJDK 64-Bit Server VM, 17.0.12, 17.0.12+7-LTS
	jvm.classpath = /opt/kafka/plugins/*:/opt/kafka/bin/../libs/accessors-smart-2.4.9.jar:/opt/kafka/bin/../libs/activation-1.1.1.jar:/opt/kafka/bin/../libs/animal-sniffer-annotations-1.23.jar:/opt/kafka/bin/../libs/annotations-4.1.1.4.jar:/opt/kafka/bin/../libs/aopalliance-repackaged-2.6.1.jar:/opt/kafka/bin/../libs/argparse4j-0.7.0.jar:/opt/kafka/bin/../libs/audience-annotations-0.12.0.jar:/opt/kafka/bin/../libs/caffeine-2.9.3.jar:/opt/kafka/bin/../libs/checker-qual-3.19.0.jar:/opt/kafka/bin/../libs/commons-beanutils-1.9.4.jar:/opt/kafka/bin/../libs/commons-cli-1.4.jar:/opt/kafka/bin/../libs/commons-collections-3.2.2.jar:/opt/kafka/bin/../libs/commons-digester-2.1.jar:/opt/kafka/bin/../libs/commons-io-2.11.0.jar:/opt/kafka/bin/../libs/commons-lang-2.6.jar:/opt/kafka/bin/../libs/commons-lang3-3.12.0.jar:/opt/kafka/bin/../libs/commons-logging-1.2.jar:/opt/kafka/bin/../libs/commons-validator-1.7.jar:/opt/kafka/bin/../libs/connect-api-3.8.0.jar:/opt/kafka/bin/../libs/connect-basic-auth-extension-3.8.0.jar:/opt/kafka/bin/../libs/connect-json-3.8.0.jar:/opt/kafka/bin/../libs/connect-mirror-3.8.0.jar:/opt/kafka/bin/../libs/connect-mirror-client-3.8.0.jar:/opt/kafka/bin/../libs/connect-runtime-3.8.0.jar:/opt/kafka/bin/../libs/connect-transforms-3.8.0.jar:/opt/kafka/bin/../libs/cruise-control-metrics-reporter-2.5.141.jar:/opt/kafka/bin/../libs/error_prone_annotations-2.10.0.jar:/opt/kafka/bin/../libs/failureaccess-1.0.1.jar:/opt/kafka/bin/../libs/grpc-api-1.61.0.jar:/opt/kafka/bin/../libs/grpc-context-1.61.0.jar:/opt/kafka/bin/../libs/grpc-core-1.61.0.jar:/opt/kafka/bin/../libs/grpc-netty-shaded-1.61.0.jar:/opt/kafka/bin/../libs/grpc-stub-1.60.1.jar:/opt/kafka/bin/../libs/grpc-util-1.61.0.jar:/opt/kafka/bin/../libs/gson-2.10.1.jar:/opt/kafka/bin/../libs/guava-32.1.3-jre.jar:/opt/kafka/bin/../libs/hk2-api-2.6.1.jar:/opt/kafka/bin/../libs/hk2-locator-2.6.1.jar:/opt/kafka/bin/../libs/hk2-utils-2.6.1.jar:/opt/kafka/bin/../libs/j2objc-annotations-2.8.jar:/opt/kafka/bin/../libs/jackson-annotations-2.16.2.jar:/opt/kafka/bin/../libs/jackson-core-2.16.2.jar:/opt/kafka/bin/../libs/jackson-databind-2.16.2.jar:/opt/kafka/bin/../libs/jackson-dataformat-csv-2.16.2.jar:/opt/kafka/bin/../libs/jackson-dataformat-yaml-2.16.2.jar:/opt/kafka/bin/../libs/jackson-datatype-jdk8-2.16.2.jar:/opt/kafka/bin/../libs/jackson-datatype-jsr310-2.16.2.jar:/opt/kafka/bin/../libs/jackson-jaxrs-base-2.16.2.jar:/opt/kafka/bin/../libs/jackson-jaxrs-json-provider-2.16.2.jar:/opt/kafka/bin/../libs/jackson-module-afterburner-2.16.2.jar:/opt/kafka/bin/../libs/jackson-module-jaxb-annotations-2.16.2.jar:/opt/kafka/bin/../libs/jackson-module-scala_2.13-2.16.2.jar:/opt/kafka/bin/../libs/jakarta.activation-api-1.2.2.jar:/opt/kafka/bin/../libs/jakarta.annotation-api-1.3.5.jar:/opt/kafka/bin/../libs/jakarta.inject-2.6.1.jar:/opt/kafka/bin/../libs/jakarta.validation-api-2.0.2.jar:/opt/kafka/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/opt/kafka/bin/../libs/jakarta.xml.bind-api-2.3.3.jar:/opt/kafka/bin/../libs/javassist-3.29.2-GA.jar:/opt/kafka/bin/../libs/javax.activation-api-1.2.0.jar:/opt/kafka/bin/../libs/javax.annotation-api-1.3.2.jar:/opt/kafka/bin/../libs/javax.servlet-api-3.1.0.jar:/opt/kafka/bin/../libs/javax.ws.rs-api-2.1.1.jar:/opt/kafka/bin/../libs/jaxb-api-2.3.1.jar:/opt/kafka/bin/../libs/jersey-client-2.39.1.jar:/opt/kafka/bin/../libs/jersey-common-2.39.1.jar:/opt/kafka/bin/../libs/jersey-container-servlet-2.39.1.jar:/opt/kafka/bin/../libs/jersey-container-servlet-core-2.39.1.jar:/opt/kafka/bin/../libs/jersey-hk2-2.39.1.jar:/opt/kafka/bin/../libs/jersey-server-2.39.1.jar:/opt/kafka/bin/../libs/jetty-client-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-continuation-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-http-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-io-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-security-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-server-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-servlet-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-servlets-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-util-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jetty-util-ajax-9.4.54.v20240208.jar:/opt/kafka/bin/../libs/jline-3.25.1.jar:/opt/kafka/bin/../libs/jmx_prometheus_javaagent-1.0.1.jar:/opt/kafka/bin/../libs/jopt-simple-5.0.4.jar:/opt/kafka/bin/../libs/jose4j-0.9.4.jar:/opt/kafka/bin/../libs/json-path-2.9.0.jar:/opt/kafka/bin/../libs/json-smart-2.4.9.jar:/opt/kafka/bin/../libs/jsonevent-layout-1.7.jar:/opt/kafka/bin/../libs/jsr305-3.0.2.jar:/opt/kafka/bin/../libs/kafka-agent-0.44.0-SNAPSHOT.jar:/opt/kafka/bin/../libs/kafka-clients-3.8.0.jar:/opt/kafka/bin/../libs/kafka-env-var-config-provider-1.1.0.jar:/opt/kafka/bin/../libs/kafka-group-coordinator-3.8.0.jar:/opt/kafka/bin/../libs/kafka-group-coordinator-api-3.8.0.jar:/opt/kafka/bin/../libs/kafka-kubernetes-config-provider-1.2.0.jar:/opt/kafka/bin/../libs/kafka-log4j-appender-3.8.0.jar:/opt/kafka/bin/../libs/kafka-metadata-3.8.0.jar:/opt/kafka/bin/../libs/kafka-oauth-client-0.15.0.jar:/opt/kafka/bin/../libs/kafka-oauth-common-0.15.0.jar:/opt/kafka/bin/../libs/kafka-oauth-keycloak-authorizer-0.15.0.jar:/opt/kafka/bin/../libs/kafka-oauth-server-0.15.0.jar:/opt/kafka/bin/../libs/kafka-oauth-server-plain-0.15.0.jar:/opt/kafka/bin/../libs/kafka-quotas-plugin-0.3.1.jar:/opt/kafka/bin/../libs/kafka-raft-3.8.0.jar:/opt/kafka/bin/../libs/kafka-server-3.8.0.jar:/opt/kafka/bin/../libs/kafka-server-common-3.8.0.jar:/opt/kafka/bin/../libs/kafka-shell-3.8.0.jar:/opt/kafka/bin/../libs/kafka-storage-3.8.0.jar:/opt/kafka/bin/../libs/kafka-storage-api-3.8.0.jar:/opt/kafka/bin/../libs/kafka-streams-3.8.0.jar:/opt/kafka/bin/../libs/kafka-streams-examples-3.8.0.jar:/opt/kafka/bin/../libs/kafka-streams-scala_2.13-3.8.0.jar:/opt/kafka/bin/../libs/kafka-streams-test-utils-3.8.0.jar:/opt/kafka/bin/../libs/kafka-tools-3.8.0.jar:/opt/kafka/bin/../libs/kafka-tools-api-3.8.0.jar:/opt/kafka/bin/../libs/kafka-transaction-coordinator-3.8.0.jar:/opt/kafka/bin/../libs/kafka_2.13-3.8.0.jar:/opt/kafka/bin/../libs/kubernetes-client-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-client-api-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-httpclient-jdk-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-model-apps-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-model-batch-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-model-certificates-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-model-common-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-model-core-6.13.4.jar:/opt/kafka/bin/../libs/kubernetes-model-extensions-6.13.4.jar:/opt/kafka/bin/../libs/listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar:/opt/kafka/bin/../libs/lz4-java-1.8.0.jar:/opt/kafka/bin/../libs/maven-artifact-3.9.6.jar:/opt/kafka/bin/../libs/metrics-core-2.2.0.jar:/opt/kafka/bin/../libs/metrics-core-4.1.12.1.jar:/opt/kafka/bin/../libs/mirror-maker-2-extensions-1.2.0.jar:/opt/kafka/bin/../libs/mirror-maker-agent-0.44.0-SNAPSHOT.jar:/opt/kafka/bin/../libs/netty-buffer-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-codec-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-common-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-handler-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-resolver-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-transport-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-transport-classes-epoll-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-transport-native-epoll-4.1.110.Final.jar:/opt/kafka/bin/../libs/netty-transport-native-unix-common-4.1.110.Final.jar:/opt/kafka/bin/../libs/nimbus-jose-jwt-9.37.2.jar:/opt/kafka/bin/../libs/opa-authorizer-1.5.1.jar:/opt/kafka/bin/../libs/opentelemetry-api-1.32.0.jar:/opt/kafka/bin/../libs/opentelemetry-api-events-1.34.1-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-context-1.32.0.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-common-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-otlp-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-otlp-common-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-sender-grpc-managed-channel-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-sender-jdk-1.34.1-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-extension-incubator-1.32.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-instrumentation-api-1.32.0.jar:/opt/kafka/bin/../libs/opentelemetry-instrumentation-api-semconv-1.32.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-kafka-clients-2.6-1.32.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-kafka-clients-common-1.32.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-proto-1.0.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-common-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-extension-autoconfigure-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-extension-autoconfigure-spi-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-logs-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-metrics-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-trace-1.34.1.jar:/opt/kafka/bin/../libs/opentelemetry-semconv-1.21.0-alpha.jar:/opt/kafka/bin/../libs/osgi-resource-locator-1.0.3.jar:/opt/kafka/bin/../libs/paranamer-2.8.jar:/opt/kafka/bin/../libs/pcollections-4.0.1.jar:/opt/kafka/bin/../libs/perfmark-api-0.26.0.jar:/opt/kafka/bin/../libs/plexus-utils-3.5.1.jar:/opt/kafka/bin/../libs/protobuf-java-3.23.4.jar:/opt/kafka/bin/../libs/reflections-0.10.2.jar:/opt/kafka/bin/../libs/reload4j-1.2.25.jar:/opt/kafka/bin/../libs/rocksdbjni-7.9.2.jar:/opt/kafka/bin/../libs/scala-collection-compat_2.13-2.10.0.jar:/opt/kafka/bin/../libs/scala-java8-compat_2.13-1.0.2.jar:/opt/kafka/bin/../libs/scala-library-2.13.14.jar:/opt/kafka/bin/../libs/scala-logging_2.13-3.9.4.jar:/opt/kafka/bin/../libs/scala-reflect-2.13.14.jar:/opt/kafka/bin/../libs/slf4j-api-1.7.36.jar:/opt/kafka/bin/../libs/slf4j-reload4j-1.7.36.jar:/opt/kafka/bin/../libs/snakeyaml-2.2.jar:/opt/kafka/bin/../libs/snakeyaml-engine-2.7.jar:/opt/kafka/bin/../libs/snappy-java-1.1.10.5.jar:/opt/kafka/bin/../libs/spotbugs-annotations-4.7.2.jar:/opt/kafka/bin/../libs/swagger-annotations-2.2.8.jar:/opt/kafka/bin/../libs/tracing-agent-0.44.0-SNAPSHOT.jar:/opt/kafka/bin/../libs/trogdor-3.8.0.jar:/opt/kafka/bin/../libs/zjsonpatch-0.3.0.jar:/opt/kafka/bin/../libs/zookeeper-3.8.4.jar:/opt/kafka/bin/../libs/zookeeper-jute-3.8.4.jar:/opt/kafka/bin/../libs/zstd-jni-1.5.6-3.jar
	os.spec = Linux, amd64, 5.14.0-284.71.1.el9_2.x86_64
	os.vcpus = 8
 (org.apache.kafka.connect.runtime.WorkerInfo) [main]
2024-10-17 09:38:52,078 INFO Scanning for plugin classes. This might take a moment ... (org.apache.kafka.connect.cli.AbstractConnectCli) [main]
2024-10-17 09:38:52,132 INFO Loading plugin from: /opt/kafka/plugins/kafka-connect-elasticsearch (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:53,753 INFO Registered loader: PluginClassLoader{pluginLocation=file:/opt/kafka/plugins/kafka-connect-elasticsearch/} (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:53,754 INFO Loading plugin from: /opt/kafka/plugins/kafka-connect-avro-converter (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:54,043 INFO Registered loader: PluginClassLoader{pluginLocation=file:/opt/kafka/plugins/kafka-connect-avro-converter/} (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:54,044 INFO Loading plugin from: /opt/kafka/plugins/kafka-connect-datagen (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:54,276 INFO Registered loader: PluginClassLoader{pluginLocation=file:/opt/kafka/plugins/kafka-connect-datagen/} (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:54,276 INFO Loading plugin from: classpath (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:55,779 INFO Registered loader: jdk.internal.loader.ClassLoaders$AppClassLoader@2b2fa4f7 (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:55,780 INFO Scanning plugins with ReflectionScanner took 3649 ms (org.apache.kafka.connect.runtime.isolation.PluginScanner) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.transforms.ReplaceField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.transforms.Filter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.transforms.InsertField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'io.strimzi.kafka.EnvVarConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,785 INFO Added plugin 'org.apache.kafka.connect.transforms.DropHeaders' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.Cast$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.InsertHeader' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.common.config.provider.DirectoryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.Flatten$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.HeaderFrom$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.SetSchemaMetadata$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.TopicNameMatches' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.ReplaceField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.SetSchemaMetadata$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.RecordIsTombstone' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'io.confluent.connect.avro.AvroConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,786 INFO Added plugin 'org.apache.kafka.connect.transforms.Cast$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.Flatten$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'io.confluent.connect.elasticsearch.ElasticsearchSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.ExtractField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampConverter$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'io.strimzi.kafka.KubernetesConfigMapConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampConverter$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.RegexRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.HoistField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.common.config.provider.FileConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'io.confluent.kafka.connect.datagen.DatagenConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'io.confluent.kafka.schemaregistry.client.config.provider.SchemaRegistryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,787 INFO Added plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.HeaderFrom$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.HasHeaderKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.MaskField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.common.config.provider.EnvVarConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.MaskField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.ExtractField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.InsertField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.converters.BooleanConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'io.strimzi.kafka.KubernetesSecretConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.transforms.HoistField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,788 INFO Added plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,790 INFO Added alias 'RecordIsTombstone' to plugin 'org.apache.kafka.connect.transforms.predicates.RecordIsTombstone' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'ElasticsearchSinkConnector' to plugin 'io.confluent.connect.elasticsearch.ElasticsearchSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'String' to plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'MirrorCheckpointConnector' to plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'Boolean' to plugin 'org.apache.kafka.connect.converters.BooleanConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'AvroConverter' to plugin 'io.confluent.connect.avro.AvroConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'KubernetesSecret' to plugin 'io.strimzi.kafka.KubernetesSecretConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'SchemaRegistryConfigProvider' to plugin 'io.confluent.kafka.schemaregistry.client.config.provider.SchemaRegistryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'Json' to plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'StringConverter' to plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'IntegerConverter' to plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'Float' to plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'LongConverter' to plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'DropHeaders' to plugin 'org.apache.kafka.connect.transforms.DropHeaders' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'SimpleHeaderConverter' to plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'DirectoryConfigProvider' to plugin 'org.apache.kafka.common.config.provider.DirectoryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'ShortConverter' to plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,791 INFO Added alias 'KubernetesSecretConfigProvider' to plugin 'io.strimzi.kafka.KubernetesSecretConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'BasicAuthSecurityRestExtension' to plugin 'org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'KubernetesConfigMapConfigProvider' to plugin 'io.strimzi.kafka.KubernetesConfigMapConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'Avro' to plugin 'io.confluent.connect.avro.AvroConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'Simple' to plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'AllConnectorClientConfigOverridePolicy' to plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'MirrorSource' to plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'Directory' to plugin 'org.apache.kafka.common.config.provider.DirectoryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'MirrorHeartbeat' to plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'BooleanConverter' to plugin 'org.apache.kafka.connect.converters.BooleanConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'HasHeaderKey' to plugin 'org.apache.kafka.connect.transforms.predicates.HasHeaderKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'MirrorCheckpoint' to plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'Datagen' to plugin 'io.confluent.kafka.connect.datagen.DatagenConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'None' to plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'TimestampRouter' to plugin 'org.apache.kafka.connect.transforms.TimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'Principal' to plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'All' to plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,792 INFO Added alias 'JsonConverter' to plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'DatagenConnector' to plugin 'io.confluent.kafka.connect.datagen.DatagenConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'RegexRouter' to plugin 'org.apache.kafka.connect.transforms.RegexRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'ByteArray' to plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'NoneConnectorClientConfigOverridePolicy' to plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'Short' to plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'Double' to plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'FileConfigProvider' to plugin 'org.apache.kafka.common.config.provider.FileConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'Long' to plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'ElasticsearchSink' to plugin 'io.confluent.connect.elasticsearch.ElasticsearchSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'File' to plugin 'org.apache.kafka.common.config.provider.FileConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'FloatConverter' to plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'ByteArrayConverter' to plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'DoubleConverter' to plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'TopicNameMatches' to plugin 'org.apache.kafka.connect.transforms.predicates.TopicNameMatches' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'MirrorHeartbeatConnector' to plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'InsertHeader' to plugin 'org.apache.kafka.connect.transforms.InsertHeader' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,793 INFO Added alias 'KubernetesConfigMap' to plugin 'io.strimzi.kafka.KubernetesConfigMapConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,794 INFO Added alias 'MirrorSourceConnector' to plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,794 INFO Added alias 'PrincipalConnectorClientConfigOverridePolicy' to plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,794 INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,794 INFO Added alias 'Integer' to plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,794 INFO Added alias 'Filter' to plugin 'org.apache.kafka.connect.transforms.Filter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2024-10-17 09:38:55,794 INFO Added alias 'SchemaRegistry' to plugin 'io.confluent.kafka.schemaregistry.client.config.provider.SchemaRegistryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
.
.
.
.
.
.
.
.
.
.
.
.
.
.
.

2024-10-17 09:40:31,710 INFO 10.7.37.158 - - [17/Oct/2024:09:40:31 +0000] "GET /connector-plugins HTTP/1.1" 200 512 "-" "-" 3 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp572980523-47]
2024-10-17 09:40:31,717 INFO 10.7.37.158 - - [17/Oct/2024:09:40:31 +0000] "GET /connectors HTTP/1.1" 200 2 "-" "-" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp572980523-48]
2024-10-17 09:40:32,546 INFO 10.7.37.158 - - [17/Oct/2024:09:40:32 +0000] "GET /connectors/datagen-source-connector/config HTTP/1.1" 404 75 "-" "-" 17 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp572980523-32]
2024-10-17 09:40:32,590 INFO DatagenConnectorConfig values: 
	iterations = 10000000
	kafka.topic = inventory
	max.interval = 1000
	quickstart = inventory
	random.seed = null
	schema.filename = 
	schema.keyfield = 
	schema.string = 
 (io.confluent.kafka.connect.datagen.DatagenConnectorConfig) [pool-2-thread-1]
2024-10-17 09:40:32,721 INFO AbstractConfig values: 
 (org.apache.kafka.common.config.AbstractConfig) [pool-2-thread-1]
2024-10-17 09:40:32,848 ERROR Error forwarding REST request (org.apache.kafka.connect.runtime.rest.RestClient) [qtp572980523-30]
org.apache.kafka.connect.runtime.rest.errors.ConnectRestException: Failed to find any class that implements Connector and which name matches io.confluent.kafka.connect.datagen.DatagenConnector, available connectors are: PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorCheckpointConnector, name='org.apache.kafka.connect.mirror.MirrorCheckpointConnector', version='3.8.0', encodedVersion=3.8.0, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorHeartbeatConnector, name='org.apache.kafka.connect.mirror.MirrorHeartbeatConnector', version='3.8.0', encodedVersion=3.8.0, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorSourceConnector, name='org.apache.kafka.connect.mirror.MirrorSourceConnector', version='3.8.0', encodedVersion=3.8.0, type=source, typeName='source', location='classpath'}
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:183)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:137)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:81)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:124)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:141)
	at org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource.putConnectorConfig(ConnectorsResource.java:233)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:569)
	at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52)
	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:134)
	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:177)
	at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$ResponseOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:176)
	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:81)
	at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:478)
	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:400)
	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81)
	at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:256)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265)
	at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:235)
	at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:684)
	at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394)
	at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:346)
	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:358)
	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:311)
	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205)
	at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:799)
	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:554)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
	at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1440)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188)
	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:505)
	at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186)
	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1355)
	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)
	at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:191)
	at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:181)
	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)
	at org.eclipse.jetty.server.Server.handle(Server.java:516)
	at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:487)
	at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:732)
	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:479)
	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277)
	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
	at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131)
	at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:409)
	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:883)
	at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1034)
	at java.base/java.lang.Thread.run(Thread.java:840)
2024-10-17 09:40:32,858 INFO 10.7.37.158 - - [17/Oct/2024:09:40:32 +0000] "PUT /connectors/datagen-source-connector/config HTTP/1.1" 500 898 "-" "-" 306 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp572980523-30]

[scholzj]

Please do not CC any people directly on your questions. People will either see your questions and be able to help or not. For example, I know nothing about this connector so I will be unable to help regardless of how much you mention me. Also, in order for anyone to be able to help:

• Make sure to properly format any input to make it readable
• Share the full logs and not just the one line that seems important to you
• Do not share any private templates of your Helm Charts. Share the actual custom resources as you created them. You can get them with kubectl get ... -o yaml

[kotapullarao]

Thank you for quick feedback. I have updated the discussion now. Please verify and help me to resolve the issue.

[katheris]

@kotapullarao I wonder if the error message is slightly misleading and that actually Connect is unable to connect to Kafka.

GET /connectors/datagen-source-connector/config succeeded but PUT /connectors/datagen-source-connector/config failed, and the PUT command also requires Connect to write to the config topic. So I wonder if it's that part that is actually failing.

It looks like the connection to Kafka is SASL_SSL, but you haven't provided any trustedCertificates. Does the Confluent Cloud bootstrap server present a certificate signed by a well known CA, or have you missed configuration trustedCertificates?

[kotapullarao]

Hi @katheris , Thanks for checking this and your response.

I have verified that if the connect cluster is able to communicate with Kafka cluster. Please find attached screenshot that confirms.
Also I am using as below and authenticated using credentials.

tls:
trustedCertificates: []
authentication:
passwordSecret:
password: password
secretName: strimzi-cloud-secret
type: plain
username: USERNAME

[katheris]

@kotapullarao I'm surprised you don't have any consumers connected to those topics. I'm not sure if the existence of the topics is enough to be sure Connect can reach Kafka. Have you had any success running any other connectors other than this one?

[kotapullarao]

Hi again @katheris , Thank you for your reply.

I am not sure why no consumers. But, I have finally managed to make this connector to work successfully with below config. I have removed the credentials and schema registry etc. (I am using credentials already for connect cluster) and converter classes stick to json and string instead of avro converter.

datagen-source-connector:
  spec:
    class: io.confluent.kafka.connect.datagen.DatagenConnector
    config:
      name: datagen-source-connector
      connector.class: io.confluent.kafka.connect.datagen.DatagenConnector
      topics: inventory
      kafka.topic: "inventory"
      quickstart: "inventory"
      max.interval: "1000"
      iterations: "10000000"
      tasks.max: "1"
      key.converter: "org.apache.kafka.connect.storage.StringConverter"
      value.converter: "org.apache.kafka.connect.json.JsonConverter"
      value.converter.schemas.enable: "false"
      type.name: kafka-connect

[katheris]

@kotapullarao I'm glad you got it working, can we close the discussion or did you have some other questions?

[kotapullarao]

@katheris , You can close the discussion. No further questions on this. Thank you.