dmathieu / kafka-connect-opensearch Goto Github PK

To smoothen the migration, all tests are currently ignoring invalid/self-signed certificates.
We need to get the generated certificate and validate it within the tests.

Kerberos configuration isn't setup in tests. We need to bring it back and reenable the tests.

Hi,

I've tried to make kafka-connect-elasticsearch work with opensearch, and happen to know this project.
I wonder this connect is support the default opensearch's docker image which is enabled the secure plugin.
The secure plugin setup the ssl but not the cert verification, so we need to add insecure option unless we setup the ssl keys.
It'll be the best if this connect support such "insecure" ssl mode.

Thank you!

I'm trying to use the kafka-connect-opensearch with kafka-connect but it keeps giving me "Failed to find any class that implements Connector and which name matches com.dmathieu.kafka.opensearch.OpenSearchSinkConnector, available connectors are: ...."
Here is the config used:

• Environment variables:
  
• docker compose service:
  
• A snapshot from the log where I see that the jar is loaded
  
• Curl command:
  
• Curl response:
  
• The full log
  Launching Kafka Connect worker ===> User uid=1000(appuser) gid=1000(appuser) groups=1000(appuser) ===> Configuring ... ===> Running preflight checks ... ===> Check if Kafka is healthy ... ===> Launching ... ===> Launching kafka-connect ... [2022-05-03 09:32:46,039] INFO WorkerInfo values: jvm.args = -Xms256M, -Xmx2G, -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=/var/log/kafka, -Dlog4j.configuration=file:/etc/kafka/connect-log4j.properties jvm.spec = Azul Systems, Inc., OpenJDK 64-Bit Server VM, 11.0.14.1, 11.0.14.1+1-LTS jvm.classpath = /etc/kafka-connect/jars/*:/usr/share/java/kafka/kafka-log4j-appender-6.2.4-ccs.jar:/usr/share/java/kafka/connect-mirror-client-6.2.4-ccs.jar:/usr/share/java/kafka/jackson-annotations-2.10.5.jar:/usr/share/java/kafka/paranamer-2.8.jar:/usr/share/java/kafka/kafka-tools-6.2.4-ccs.jar:/usr/share/java/kafka/jackson-module-scala_2.13-2.10.5.jar:/usr/share/java/kafka/jersey-hk2-2.34.jar:/usr/share/java/kafka/jakarta.ws.rs-api-2.1.6.jar:/usr/share/java/kafka/osgi-resource-locator-1.0.3.jar:/usr/share/java/kafka/netty-tcnative-classes-2.0.46.Final.jar:/usr/share/java/kafka/aopalliance-repackaged-2.6.1.jar:/usr/share/java/kafka/metrics-core-2.2.0.jar:/usr/share/java/kafka/audience-annotations-0.5.0.jar:/usr/share/java/kafka/zstd-jni-1.4.9-1.jar:/usr/share/java/kafka/kafka-streams-scala_2.13-6.2.4-ccs.jar:/usr/share/java/kafka/kafka_2.13-6.2.4-ccs-javadoc.jar:/usr/share/java/kafka/jetty-client-9.4.44.v20210927.jar:/usr/share/java/kafka/jetty-servlets-9.4.44.v20210927.jar:/usr/share/java/kafka/confluent-log4j-1.2.17-cp2.2.jar:/usr/share/java/kafka/jackson-dataformat-csv-2.10.5.jar:/usr/share/java/kafka/javax.ws.rs-api-2.1.1.jar:/usr/share/java/kafka/jetty-util-9.4.44.v20210927.jar:/usr/share/java/kafka/rocksdbjni-5.18.4.jar:/usr/share/java/kafka/netty-buffer-4.1.73.Final.jar:/usr/share/java/kafka/jersey-common-2.34.jar:/usr/share/java/kafka/jackson-jaxrs-json-provider-2.10.5.jar:/usr/share/java/kafka/scala-library-2.13.5.jar:/usr/share/java/kafka/kafka_2.13-6.2.4-ccs-test.jar:/usr/share/java/kafka/kafka-raft-6.2.4-ccs.jar:/usr/share/java/kafka/kafka-shell-6.2.4-ccs.jar:/usr/share/java/kafka/jakarta.xml.bind-api-2.3.2.jar:/usr/share/java/kafka/connect-mirror-6.2.4-ccs.jar:/usr/share/java/kafka/scala-logging_2.13-3.9.2.jar:/usr/share/java/kafka/jetty-continuation-9.4.44.v20210927.jar:/usr/share/java/kafka/jackson-databind-2.10.5.1.jar:/usr/share/java/kafka/jetty-security-9.4.44.v20210927.jar:/usr/share/java/kafka/javax.servlet-api-3.1.0.jar:/usr/share/java/kafka/connect-basic-auth-extension-6.2.4-ccs.jar:/usr/share/java/kafka/kafka_2.13-6.2.4-ccs.jar:/usr/share/java/kafka/netty-resolver-4.1.73.Final.jar:/usr/share/java/kafka/activation-1.1.1.jar:/usr/share/java/kafka/scala-collection-compat_2.13-2.3.0.jar:/usr/share/java/kafka/hk2-utils-2.6.1.jar:/usr/share/java/kafka/snappy-java-1.1.8.1.jar:/usr/share/java/kafka/hk2-api-2.6.1.jar:/usr/share/java/kafka/hk2-locator-2.6.1.jar:/usr/share/java/kafka/jackson-core-2.10.5.jar:/usr/share/java/kafka/connect-runtime-6.2.4-ccs.jar:/usr/share/java/kafka/netty-codec-4.1.73.Final.jar:/usr/share/java/kafka/netty-transport-native-epoll-4.1.73.Final.jar:/usr/share/java/kafka/commons-lang3-3.8.1.jar:/usr/share/java/kafka/jakarta.validation-api-2.0.2.jar:/usr/share/java/kafka/jetty-server-9.4.44.v20210927.jar:/usr/share/java/kafka/jakarta.annotation-api-1.3.5.jar:/usr/share/java/kafka/kafka-streams-test-utils-6.2.4-ccs.jar:/usr/share/java/kafka/connect-transforms-6.2.4-ccs.jar:/usr/share/java/kafka/reflections-0.9.12.jar:/usr/share/java/kafka/zookeeper-3.5.9.jar:/usr/share/java/kafka/connect-json-6.2.4-ccs.jar:/usr/share/java/kafka/connect-api-6.2.4-ccs.jar:/usr/share/java/kafka/jopt-simple-5.0.4.jar:/usr/share/java/kafka/jackson-datatype-jdk8-2.10.5.jar:/usr/share/java/kafka/jersey-server-2.34.jar:/usr/share/java/kafka/commons-cli-1.4.jar:/usr/share/java/kafka/jakarta.activation-api-1.2.1.jar:/usr/share/java/kafka/netty-transport-classes-epoll-4.1.73.Final.jar:/usr/share/java/kafka/jetty-http-9.4.44.v20210927.jar:/usr/share/java/kafka/netty-handler-4.1.73.Final.jar:/usr/share/java/kafka/jetty-io-9.4.44.v20210927.jar:/usr/share/java/kafka/argparse4j-0.7.0.jar:/usr/share/java/kafka/netty-transport-4.1.73.Final.jar:/usr/share/java/kafka/kafka-streams-examples-6.2.4-ccs.jar:/usr/share/java/kafka/slf4j-api-1.7.30.jar:/usr/share/java/kafka/jackson-module-paranamer-2.10.5.jar:/usr/share/java/kafka/kafka_2.13-6.2.4-ccs-sources.jar:/usr/share/java/kafka/kafka_2.13-6.2.4-ccs-test-sources.jar:/usr/share/java/kafka/jakarta.inject-2.6.1.jar:/usr/share/java/kafka/lz4-java-1.7.1.jar:/usr/share/java/kafka/kafka-metadata-6.2.4-ccs.jar:/usr/share/java/kafka/jersey-client-2.34.jar:/usr/share/java/kafka/scala-java8-compat_2.13-0.9.1.jar:/usr/share/java/kafka/jline-3.12.1.jar:/usr/share/java/kafka/jackson-jaxrs-base-2.10.5.jar:/usr/share/java/kafka/plexus-utils-3.2.1.jar:/usr/share/java/kafka/jetty-util-ajax-9.4.44.v20210927.jar:/usr/share/java/kafka/kafka.jar:/usr/share/java/kafka/jersey-container-servlet-core-2.34.jar:/usr/share/java/kafka/jaxb-api-2.3.0.jar:/usr/share/java/kafka/kafka-clients-6.2.4-ccs.jar:/usr/share/java/kafka/scala-reflect-2.13.5.jar:/usr/share/java/kafka/maven-artifact-3.8.1.jar:/usr/share/java/kafka/zookeeper-jute-3.5.9.jar:/usr/share/java/kafka/netty-common-4.1.73.Final.jar:/usr/share/java/kafka/javassist-3.27.0-GA.jar:/usr/share/java/kafka/slf4j-log4j12-1.7.30.jar:/usr/share/java/kafka/kafka-streams-6.2.4-ccs.jar:/usr/share/java/kafka/netty-transport-native-unix-common-4.1.73.Final.jar:/usr/share/java/kafka/jersey-container-servlet-2.34.jar:/usr/share/java/kafka/jackson-module-jaxb-annotations-2.10.5.jar:/usr/share/java/kafka/jetty-servlet-9.4.44.v20210927.jar:/usr/share/java/confluent-common/common-config-6.2.4.jar:/usr/share/java/confluent-common/build-tools-6.2.4.jar:/usr/share/java/confluent-common/slf4j-api-1.7.30.jar:/usr/share/java/confluent-common/common-utils-6.2.4.jar:/usr/share/java/confluent-common/common-metrics-6.2.4.jar:/usr/share/java/kafka-serde-tools/jackson-annotations-2.10.5.jar:/usr/share/java/kafka-serde-tools/kafka-json-schema-serializer-6.2.4.jar:/usr/share/java/kafka-serde-tools/joda-time-2.9.9.jar:/usr/share/java/kafka-serde-tools/failureaccess-1.0.1.jar:/usr/share/java/kafka-serde-tools/grpc-core-0.15.0.jar:/usr/share/java/kafka-serde-tools/avro-1.10.1.jar:/usr/share/java/kafka-serde-tools/classgraph-4.8.21.jar:/usr/share/java/kafka-serde-tools/kafka-streams-avro-serde-6.2.4.jar:/usr/share/java/kafka-serde-tools/kafka-json-schema-provider-6.2.4.jar:/usr/share/java/kafka-serde-tools/mbknor-jackson-jsonschema_2.13-1.0.39.jar:/usr/share/java/kafka-serde-tools/error_prone_annotations-2.5.1.jar:/usr/share/java/kafka-serde-tools/kotlin-stdlib-1.4.21.jar:/usr/share/java/kafka-serde-tools/jackson-dataformat-yaml-2.10.5.jar:/usr/share/java/kafka-serde-tools/swagger-models-1.6.2.jar:/usr/share/java/kafka-serde-tools/rocksdbjni-5.18.4.jar:/usr/share/java/kafka-serde-tools/commons-collections-3.2.2.jar:/usr/share/java/kafka-serde-tools/checker-qual-3.8.0.jar:/usr/share/java/kafka-serde-tools/scala-library-2.13.5.jar:/usr/share/java/kafka-serde-tools/googleapis-common-protos-0.0.3.jar:/usr/share/java/kafka-serde-tools/kafka-streams-json-schema-serde-6.2.4.jar:/usr/share/java/kafka-serde-tools/kotlinx-coroutines-core-1.3.7.jar:/usr/share/java/kafka-serde-tools/kotlin-stdlib-jdk7-1.4.10.jar:/usr/share/java/kafka-serde-tools/kotlin-scripting-common-1.4.21.jar:/usr/share/java/kafka-serde-tools/grpc-protobuf-lite-0.15.0.jar:/usr/share/java/kafka-serde-tools/swagger-annotations-1.6.2.jar:/usr/share/java/kafka-serde-tools/jackson-datatype-joda-2.10.5.jar:/usr/share/java/kafka-serde-tools/jackson-databind-2.10.5.1.jar:/usr/share/java/kafka-serde-tools/gson-2.8.6.jar:/usr/share/java/kafka-serde-tools/org.everit.json.schema-1.12.2.jar:/usr/share/java/kafka-serde-tools/kotlin-scripting-compiler-impl-embeddable-1.3.50.jar:/usr/share/java/kafka-serde-tools/jackson-core-2.10.5.jar:/usr/share/java/kafka-serde-tools/guava-30.1.1-jre.jar:/usr/share/java/kafka-serde-tools/commons-lang3-3.8.1.jar:/usr/share/java/kafka-serde-tools/commons-logging-1.2.jar:/usr/share/java/kafka-serde-tools/kotlin-scripting-jvm-1.4.21.jar:/usr/share/java/kafka-serde-tools/commons-validator-1.6.jar:/usr/share/java/kafka-serde-tools/kafka-schema-registry-client-6.2.4.jar:/usr/share/java/kafka-serde-tools/re2j-1.3.jar:/usr/share/java/kafka-serde-tools/kotlin-script-runtime-1.4.21.jar:/usr/share/java/kafka-serde-tools/listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar:/usr/share/java/kafka-serde-tools/jackson-datatype-jsr310-2.10.5.jar:/usr/share/java/kafka-serde-tools/protobuf-java-3.11.4.jar:/usr/share/java/kafka-serde-tools/annotations-13.0.jar:/usr/share/java/kafka-serde-tools/protobuf-java-util-3.11.4.jar:/usr/share/java/kafka-serde-tools/kafka-protobuf-types-6.2.4.jar:/usr/share/java/kafka-serde-tools/grpc-stub-0.15.0.jar:/usr/share/java/kafka-serde-tools/kafka-protobuf-provider-6.2.4.jar:/usr/share/java/kafka-serde-tools/kafka-streams-protobuf-serde-6.2.4.jar:/usr/share/java/kafka-serde-tools/handy-uri-templates-2.1.8.jar:/usr/share/java/kafka-serde-tools/jackson-datatype-jdk8-2.10.5.jar:/usr/share/java/kafka-serde-tools/wire-schema-3.6.0.jar:/usr/share/java/kafka-serde-tools/kafka-connect-json-schema-converter-6.2.4.jar:/usr/share/java/kafka-serde-tools/okio-2.8.0.jar:/usr/share/java/kafka-serde-tools/validation-api-2.0.1.Final.jar:/usr/share/java/kafka-serde-tools/kotlin-stdlib-jdk8-1.4.10.jar:/usr/share/java/kafka-serde-tools/wire-runtime-3.6.0.jar:/usr/share/java/kafka-serde-tools/kafka-connect-avro-data-6.2.4.jar:/usr/share/java/kafka-serde-tools/kotlin-scripting-compiler-embeddable-1.3.50.jar:/usr/share/java/kafka-serde-tools/slf4j-api-1.7.30.jar:/usr/share/java/kafka-serde-tools/kafka-schema-serializer-6.2.4.jar:/usr/share/java/kafka-serde-tools/json-20201115.jar:/usr/share/java/kafka-serde-tools/kafka-json-serializer-6.2.4.jar:/usr/share/java/kafka-serde-tools/kafka-protobuf-serializer-6.2.4.jar:/usr/share/java/kafka-serde-tools/kafka-connect-avro-converter-6.2.4.jar:/usr/share/java/kafka-serde-tools/commons-digester-1.8.1.jar:/usr/share/java/kafka-serde-tools/snakeyaml-1.26.jar:/usr/share/java/kafka-serde-tools/kafka-connect-protobuf-converter-6.2.4.jar:/usr/share/java/kafka-serde-tools/kafka-avro-serializer-6.2.4.jar:/usr/share/java/kafka-serde-tools/j2objc-annotations-1.3.jar:/usr/share/java/kafka-serde-tools/jackson-datatype-guava-2.10.5.jar:/usr/share/java/kafka-serde-tools/kotlin-stdlib-common-1.4.10.jar:/usr/share/java/kafka-serde-tools/commons-compress-1.21.jar:/usr/share/java/kafka-serde-tools/kafka-streams-6.2.4-ccs.jar:/usr/share/java/kafka-serde-tools/jsr305-3.0.2.jar:/usr/share/java/kafka-serde-tools/grpc-protobuf-0.15.0.jar:/usr/share/java/kafka-serde-tools/jackson-module-parameter-names-2.10.5.jar:/usr/share/java/kafka-serde-tools/swagger-core-1.6.2.jar:/usr/share/java/monitoring-interceptors/monitoring-interceptors-6.2.4.jar:/usr/bin/../share/java/kafka/kafka-log4j-appender-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/connect-mirror-client-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/jackson-annotations-2.10.5.jar:/usr/bin/../share/java/kafka/paranamer-2.8.jar:/usr/bin/../share/java/kafka/kafka-tools-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/jackson-module-scala_2.13-2.10.5.jar:/usr/bin/../share/java/kafka/jersey-hk2-2.34.jar:/usr/bin/../share/java/kafka/jakarta.ws.rs-api-2.1.6.jar:/usr/bin/../share/java/kafka/osgi-resource-locator-1.0.3.jar:/usr/bin/../share/java/kafka/netty-tcnative-classes-2.0.46.Final.jar:/usr/bin/../share/java/kafka/aopalliance-repackaged-2.6.1.jar:/usr/bin/../share/java/kafka/metrics-core-2.2.0.jar:/usr/bin/../share/java/kafka/audience-annotations-0.5.0.jar:/usr/bin/../share/java/kafka/zstd-jni-1.4.9-1.jar:/usr/bin/../share/java/kafka/kafka-streams-scala_2.13-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/kafka_2.13-6.2.4-ccs-javadoc.jar:/usr/bin/../share/java/kafka/jetty-client-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/jetty-servlets-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/confluent-log4j-1.2.17-cp2.2.jar:/usr/bin/../share/java/kafka/jackson-dataformat-csv-2.10.5.jar:/usr/bin/../share/java/kafka/javax.ws.rs-api-2.1.1.jar:/usr/bin/../share/java/kafka/jetty-util-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/rocksdbjni-5.18.4.jar:/usr/bin/../share/java/kafka/netty-buffer-4.1.73.Final.jar:/usr/bin/../share/java/kafka/jersey-common-2.34.jar:/usr/bin/../share/java/kafka/jackson-jaxrs-json-provider-2.10.5.jar:/usr/bin/../share/java/kafka/scala-library-2.13.5.jar:/usr/bin/../share/java/kafka/kafka_2.13-6.2.4-ccs-test.jar:/usr/bin/../share/java/kafka/kafka-raft-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/kafka-shell-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/jakarta.xml.bind-api-2.3.2.jar:/usr/bin/../share/java/kafka/connect-mirror-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/scala-logging_2.13-3.9.2.jar:/usr/bin/../share/java/kafka/jetty-continuation-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/jackson-databind-2.10.5.1.jar:/usr/bin/../share/java/kafka/jetty-security-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/javax.servlet-api-3.1.0.jar:/usr/bin/../share/java/kafka/connect-basic-auth-extension-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/kafka_2.13-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/netty-resolver-4.1.73.Final.jar:/usr/bin/../share/java/kafka/activation-1.1.1.jar:/usr/bin/../share/java/kafka/scala-collection-compat_2.13-2.3.0.jar:/usr/bin/../share/java/kafka/hk2-utils-2.6.1.jar:/usr/bin/../share/java/kafka/snappy-java-1.1.8.1.jar:/usr/bin/../share/java/kafka/hk2-api-2.6.1.jar:/usr/bin/../share/java/kafka/hk2-locator-2.6.1.jar:/usr/bin/../share/java/kafka/jackson-core-2.10.5.jar:/usr/bin/../share/java/kafka/connect-runtime-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/netty-codec-4.1.73.Final.jar:/usr/bin/../share/java/kafka/netty-transport-native-epoll-4.1.73.Final.jar:/usr/bin/../share/java/kafka/commons-lang3-3.8.1.jar:/usr/bin/../share/java/kafka/jakarta.validation-api-2.0.2.jar:/usr/bin/../share/java/kafka/jetty-server-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/jakarta.annotation-api-1.3.5.jar:/usr/bin/../share/java/kafka/kafka-streams-test-utils-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/connect-transforms-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/reflections-0.9.12.jar:/usr/bin/../share/java/kafka/zookeeper-3.5.9.jar:/usr/bin/../share/java/kafka/connect-json-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/connect-api-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/jopt-simple-5.0.4.jar:/usr/bin/../share/java/kafka/jackson-datatype-jdk8-2.10.5.jar:/usr/bin/../share/java/kafka/jersey-server-2.34.jar:/usr/bin/../share/java/kafka/commons-cli-1.4.jar:/usr/bin/../share/java/kafka/jakarta.activation-api-1.2.1.jar:/usr/bin/../share/java/kafka/netty-transport-classes-epoll-4.1.73.Final.jar:/usr/bin/../share/java/kafka/jetty-http-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/netty-handler-4.1.73.Final.jar:/usr/bin/../share/java/kafka/jetty-io-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/argparse4j-0.7.0.jar:/usr/bin/../share/java/kafka/netty-transport-4.1.73.Final.jar:/usr/bin/../share/java/kafka/kafka-streams-examples-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/slf4j-api-1.7.30.jar:/usr/bin/../share/java/kafka/jackson-module-paranamer-2.10.5.jar:/usr/bin/../share/java/kafka/kafka_2.13-6.2.4-ccs-sources.jar:/usr/bin/../share/java/kafka/kafka_2.13-6.2.4-ccs-test-sources.jar:/usr/bin/../share/java/kafka/jakarta.inject-2.6.1.jar:/usr/bin/../share/java/kafka/lz4-java-1.7.1.jar:/usr/bin/../share/java/kafka/kafka-metadata-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/jersey-client-2.34.jar:/usr/bin/../share/java/kafka/scala-java8-compat_2.13-0.9.1.jar:/usr/bin/../share/java/kafka/jline-3.12.1.jar:/usr/bin/../share/java/kafka/jackson-jaxrs-base-2.10.5.jar:/usr/bin/../share/java/kafka/plexus-utils-3.2.1.jar:/usr/bin/../share/java/kafka/jetty-util-ajax-9.4.44.v20210927.jar:/usr/bin/../share/java/kafka/kafka.jar:/usr/bin/../share/java/kafka/jersey-container-servlet-core-2.34.jar:/usr/bin/../share/java/kafka/jaxb-api-2.3.0.jar:/usr/bin/../share/java/kafka/kafka-clients-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/scala-reflect-2.13.5.jar:/usr/bin/../share/java/kafka/maven-artifact-3.8.1.jar:/usr/bin/../share/java/kafka/zookeeper-jute-3.5.9.jar:/usr/bin/../share/java/kafka/netty-common-4.1.73.Final.jar:/usr/bin/../share/java/kafka/javassist-3.27.0-GA.jar:/usr/bin/../share/java/kafka/slf4j-log4j12-1.7.30.jar:/usr/bin/../share/java/kafka/kafka-streams-6.2.4-ccs.jar:/usr/bin/../share/java/kafka/netty-transport-native-unix-common-4.1.73.Final.jar:/usr/bin/../share/java/kafka/jersey-container-servlet-2.34.jar:/usr/bin/../share/java/kafka/jackson-module-jaxb-annotations-2.10.5.jar:/usr/bin/../share/java/kafka/jetty-servlet-9.4.44.v20210927.jar:/usr/bin/../share/java/confluent-telemetry/confluent-metrics-6.2.4-ce.jar os.spec = Linux, amd64, 4.19.0-8-amd64 os.vcpus = 2 (org.apache.kafka.connect.runtime.WorkerInfo) [2022-05-03 09:32:46,043] INFO Scanning for plugin classes. This might take a moment ... (org.apache.kafka.connect.cli.ConnectDistributed) [2022-05-03 09:32:46,073] INFO Loading plugin from: /usr/share/java/cp-base-new (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,727] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/cp-base-new/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,727] INFO Added plugin 'org.apache.kafka.common.config.provider.FileConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,728] INFO Added plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,728] INFO Added plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,728] INFO Added plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,731] INFO Loading plugin from: /usr/share/java/kafka-connect-opensearch.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,753] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/kafka-connect-opensearch.jar/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:48,755] INFO Loading plugin from: /usr/share/java/confluent-control-center (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,290] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/confluent-control-center/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,290] INFO Added plugin 'org.apache.kafka.connect.tools.VerifiableSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.tools.VerifiableSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.tools.MockSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.tools.MockConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.tools.MockSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.tools.SchemaSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'io.confluent.connect.avro.AvroConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'io.confluent.connect.json.JsonSchemaConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'io.confluent.connect.protobuf.ProtobufConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,291] INFO Added plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.ReplaceField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.ReplaceField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.SetSchemaMetadata$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.Filter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.InsertField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampConverter$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.MaskField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.RegexRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.HoistField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.MaskField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.Cast$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.Cast$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.runtime.PredicatedTransformation' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.ExtractField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.Flatten$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.InsertField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,292] INFO Added plugin 'org.apache.kafka.connect.transforms.Flatten$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.SetSchemaMetadata$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.ExtractField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampConverter$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.HoistField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.HasHeaderKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.RecordIsTombstone' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.TopicNameMatches' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'io.confluent.kafka.secretregistry.client.config.provider.SecretConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Added plugin 'io.confluent.connect.security.ConnectSecurityExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:58,293] INFO Loading plugin from: /usr/share/java/confluent-telemetry (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:59,335] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/confluent-telemetry/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:59,336] INFO Loading plugin from: /usr/share/java/confluent-common (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:59,346] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/confluent-common/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:32:59,347] INFO Loading plugin from: /usr/share/java/kafka-serde-tools (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:00,303] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/kafka-serde-tools/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:00,303] INFO Loading plugin from: /usr/share/java/confluent-hub-client (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:00,696] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/confluent-hub-client/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:00,697] INFO Loading plugin from: /usr/share/java/rest-utils (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:01,974] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/rest-utils/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:01,975] INFO Loading plugin from: /usr/share/java/monitoring-interceptors (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:02,290] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/monitoring-interceptors/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:02,291] INFO Loading plugin from: /usr/share/java/schema-registry (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:04,567] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/schema-registry/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:04,567] INFO Loading plugin from: /usr/share/java/acl (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:11,050] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/acl/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:11,051] INFO Loading plugin from: /usr/share/java/kafka (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:12,768] INFO Registered loader: PluginClassLoader{pluginLocation=file:/usr/share/java/kafka/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:12,768] INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:12,768] INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:12,768] INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:12,768] INFO Added plugin 'org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,114] INFO Registered loader: jdk.internal.loader.ClassLoaders$AppClassLoader@2c13da15 (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,115] INFO Added aliases 'MirrorCheckpointConnector' and 'MirrorCheckpoint' to plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'MirrorHeartbeatConnector' and 'MirrorHeartbeat' to plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'MirrorSourceConnector' and 'MirrorSource' to plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'MockConnector' and 'Mock' to plugin 'org.apache.kafka.connect.tools.MockConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'MockSinkConnector' and 'MockSink' to plugin 'org.apache.kafka.connect.tools.MockSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'MockSourceConnector' and 'MockSource' to plugin 'org.apache.kafka.connect.tools.MockSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'SchemaSourceConnector' and 'SchemaSource' to plugin 'org.apache.kafka.connect.tools.SchemaSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'VerifiableSinkConnector' and 'VerifiableSink' to plugin 'org.apache.kafka.connect.tools.VerifiableSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'VerifiableSourceConnector' and 'VerifiableSource' to plugin 'org.apache.kafka.connect.tools.VerifiableSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'AvroConverter' and 'Avro' to plugin 'io.confluent.connect.avro.AvroConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'JsonSchemaConverter' and 'JsonSchema' to plugin 'io.confluent.connect.json.JsonSchemaConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'ProtobufConverter' and 'Protobuf' to plugin 'io.confluent.connect.protobuf.ProtobufConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,116] INFO Added aliases 'ByteArrayConverter' and 'ByteArray' to plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'DoubleConverter' and 'Double' to plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'FloatConverter' and 'Float' to plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'IntegerConverter' and 'Integer' to plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'LongConverter' and 'Long' to plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'ShortConverter' and 'Short' to plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'JsonConverter' and 'Json' to plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'StringConverter' and 'String' to plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'ByteArrayConverter' and 'ByteArray' to plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'DoubleConverter' and 'Double' to plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'FloatConverter' and 'Float' to plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'IntegerConverter' and 'Integer' to plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,117] INFO Added aliases 'LongConverter' and 'Long' to plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,118] INFO Added aliases 'ShortConverter' and 'Short' to plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,118] INFO Added aliases 'JsonConverter' and 'Json' to plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,118] INFO Added alias 'SimpleHeaderConverter' to plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,118] INFO Added aliases 'StringConverter' and 'String' to plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,118] INFO Added aliases 'PredicatedTransformation' and 'Predicated' to plugin 'org.apache.kafka.connect.runtime.PredicatedTransformation' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,118] INFO Added alias 'Filter' to plugin 'org.apache.kafka.connect.transforms.Filter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'RegexRouter' to plugin 'org.apache.kafka.connect.transforms.RegexRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'TimestampRouter' to plugin 'org.apache.kafka.connect.transforms.TimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'HasHeaderKey' to plugin 'org.apache.kafka.connect.transforms.predicates.HasHeaderKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'RecordIsTombstone' to plugin 'org.apache.kafka.connect.transforms.predicates.RecordIsTombstone' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'TopicNameMatches' to plugin 'org.apache.kafka.connect.transforms.predicates.TopicNameMatches' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'ConnectSecurityExtension' to plugin 'io.confluent.connect.security.ConnectSecurityExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added alias 'BasicAuthSecurityRestExtension' to plugin 'org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added aliases 'AllConnectorClientConfigOverridePolicy' and 'All' to plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added aliases 'NoneConnectorClientConfigOverridePolicy' and 'None' to plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,119] INFO Added aliases 'PrincipalConnectorClientConfigOverridePolicy' and 'Principal' to plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [2022-05-03 09:33:17,195] INFO DistributedConfig values: access.control.allow.methods = access.control.allow.origin = admin.listeners = null bootstrap.servers = [kafka-01:9092, kafka-02:9092, kafka-03:9092] client.dns.lookup = use_all_dns_ips client.id = config.providers = [] config.storage.replication.factor = 3 config.storage.topic = docker-connect-configs connect.protocol = sessioned connections.max.idle.ms = 540000 connector.client.config.override.policy = None group.id = connect-kafka-to-elk header.converter = class org.apache.kafka.connect.storage.SimpleHeaderConverter heartbeat.interval.ms = 3000 inter.worker.key.generation.algorithm = HmacSHA256 inter.worker.key.size = null inter.worker.key.ttl.ms = 3600000 inter.worker.signature.algorithm = HmacSHA256 inter.worker.verification.algorithms = [HmacSHA256] internal.key.converter = class org.apache.kafka.connect.storage.StringConverter internal.value.converter = class org.apache.kafka.connect.json.JsonConverter key.converter = class org.apache.kafka.connect.storage.StringConverter listeners = null metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 offset.flush.interval.ms = 60000 offset.flush.timeout.ms = 5000 offset.storage.partitions = 25 offset.storage.replication.factor = 3 offset.storage.topic = docker-connect-offsets plugin.path = [/usr/share/java, /usr/share/java/kafka-connect-opensearch.jar] rebalance.timeout.ms = 60000 receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 40000 response.http.headers.config = rest.advertised.host.name = connect rest.advertised.listener = null rest.advertised.port = null rest.extension.classes = [] rest.host.name = null rest.port = 8082 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI scheduled.rebalance.max.delay.ms = 300000 security.protocol = PLAINTEXT send.buffer.bytes = 131072 session.timeout.ms = 10000 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.client.auth = none ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS status.storage.partitions = 5 status.storage.replication.factor = 3 status.storage.topic = docker-connect-status task.shutdown.graceful.timeout.ms = 5000 topic.creation.enable = true topic.tracking.allow.reset = true topic.tracking.enable = true value.converter = class org.apache.kafka.connect.json.JsonConverter worker.sync.timeout.ms = 3000 worker.unsync.backoff.ms = 300000 (org.apache.kafka.connect.runtime.distributed.DistributedConfig)

Can some please help with this issue ?

I placed kafka-connect-opensearch.jar into usr/share/java folder as mentioned in the README, but when I'm trying to create connector (via post request) - it gives me an error:

{ "error_code": 500, "message": "Failed to find any class that implements Connector and which name matches com.dmathieu.kafka.opensearch.OpenSearchSinkConnector, available connectors are: PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorCheckpointConnector, name='org.apache.kafka.connect.mirror.MirrorCheckpointConnector', version='1', encodedVersion=1, type=source, typeName='source', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorHeartbeatConnector, name='org.apache.kafka.connect.mirror.MirrorHeartbeatConnector', version='1', encodedVersion=1, type=source, typeName='source', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.mirror.MirrorSourceConnector, name='org.apache.kafka.connect.mirror.MirrorSourceConnector', version='1', encodedVersion=1, type=source, typeName='source', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockConnector, name='org.apache.kafka.connect.tools.MockConnector', version='6.2.1-ccs', encodedVersion=6.2.1-ccs, type=connector, typeName='connector', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockSinkConnector, name='org.apache.kafka.connect.tools.MockSinkConnector', version='6.2.1-ccs', encodedVersion=6.2.1-ccs, type=sink, typeName='sink', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockSourceConnector, name='org.apache.kafka.connect.tools.MockSourceConnector', version='6.2.1-ccs', encodedVersion=6.2.1-ccs, type=source, typeName='source', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.SchemaSourceConnector, name='org.apache.kafka.connect.tools.SchemaSourceConnector', version='6.2.1-ccs', encodedVersion=6.2.1-ccs, type=source, typeName='source', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.VerifiableSinkConnector, name='org.apache.kafka.connect.tools.VerifiableSinkConnector', version='6.2.1-ccs', encodedVersion=6.2.1-ccs, type=source, typeName='source', location='file:/usr/share/java/kafka/'}, PluginDesc{klass=class org.apache.kafka.connect.tools.VerifiableSourceConnector, name='org.apache.kafka.connect.tools.VerifiableSourceConnector', version='6.2.1-ccs', encodedVersion=6.2.1-ccs, type=source, typeName='source', location='file:/usr/share/java/kafka/'}" }

I installed kafka-connect with following configs
docker run -d \ --name=kafka-connect-avro \ --net=host \ -e CONNECT_BOOTSTRAP_SERVERS=localhost:9092 \ -e CONNECT_REST_PORT=8083 \ -e CONNECT_GROUP_ID="quickstart-avro" \ -e CONNECT_CONFIG_STORAGE_TOPIC="quickstart-avro-config" \ -e CONNECT_OFFSET_STORAGE_TOPIC="quickstart-avro-offsets" \ -e CONNECT_STATUS_STORAGE_TOPIC="quickstart-avro-status" \ -e CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1 \ -e CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1 \ -e CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1 \ -e CONNECT_KEY_CONVERTER="io.confluent.connect.avro.AvroConverter" \ -e CONNECT_VALUE_CONVERTER="io.confluent.connect.avro.AvroConverter" \ -e CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL="http://localhost:8081" \ -e CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL="http://localhost:8081" \ -e CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \ -e CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \ -e CONNECT_REST_ADVERTISED_HOST_NAME="localhost" \ -e CONNECT_LOG4J_ROOT_LOGLEVEL=DEBUG \ -e CONNECT_PLUGIN_PATH=/usr/share/java,/etc/kafka-connect/jars \ -v /tmp/quickstart/file:/tmp/quickstart \ -v /tmp/quickstart/jars:/etc/kafka-connect/jars \ confluentinc/cp-kafka-connect:latest

Also, plugin path includes 'usr/share/java' where I have moved jar file.

Can some please help with this issue?

1. I have configured the SASL_PLAINTEXT.

2. Updated connect-distributed.properties file -

bootstrap.servers=10.30.1.101:9092
group.id=connect-cluster

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

offset.storage.topic=connect-offsets
offset.storage.replication.factor=1

config.storage.topic=connect-configs
config.storage.replication.factor=1

status.storage.topic=connect-status
status.storage.replication.factor=1

offset.flush.interval.ms=10000

listeners=HTTP://:8083

plugin.path=/home/devuser/connectors

sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="admin-secret";

3. plugin.path = /home/devuser/connectors contains the kafka-connect-opensearch JAR file.

4. Once done, I was successfully able to start the kafka-connect server on 8083 Port.

5. Now, Using POSTMAN I tried to create my first connector.

REQUEST - POST - 10.30.1.101:8083/connectors/
JSON BODY -

{
    "name": "first-opensearch-connector",
    "config": {
    "connector.class": "com.dmathieu.kafka.opensearch.OpenSearchSinkConnector",
    "type.name": "_doc",
    "connection.password": "admin",
    "tasks.max": "1",
    "topics": "my_topic",
    "name": "first-opensearch-connector",
    "connection.username": "admin",
    "value.converter.schemas.enable": "false",
    "connection.url": "http://10.30.1.101:9200",
    "key.ignore": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "schema.ignore": "true",
    "drop.invalid.message":"true",
    "behavior.on.malformed.documents":"fail",
    "write.method":"UPSERT",
    "read.timeout.ms":"10000",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "errors.tolerance": "all",
    "transforms":"AddPrefix,TimestampRouter,InsertField",
    "transforms.AddPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex":".*",
    "transforms.AddPrefix.replacement":"acme_$0",
    "transforms.TimestampRouter.type": "org.apache.kafka.connect.transforms.TimestampRouter",
    "transforms.TimestampRouter.topic.format": "foo-bar-${topic}-${timestamp}",
    "transforms.TimestampRouter.timestamp.format": "YYYYMMdd",
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertField.static.field": "MessageSource",
    "transforms.InsertField.static.value": "Kafka Connect framework",
    "connection-compression": "gzip",
    "sasl.mechanism": "PLAIN",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required",
    "username": "admin", 
    "password": "admin-secret"}
}

Snapshots -

All Data Stream tests are currently disabled. We need to fix them and reenable them.

I'm creating the connector with the following configs but getting an authentication error:

{
    "error_code": 400,
    "message": "Connector configuration is invalid and contains the following 3 error(s):\nCould not connect to OpenSearch. Error message: Connection is closed\nCould not authenticate the user. Check the 'connection.username' and 'connection.password'. Error message: Connection is closed\nCould not authenticate the user. Check the 'connection.username' and 'connection.password'. Error message: Connection is closed\nYou can also find the above list of errors at the endpoint /connector-plugins/{connectorType}/config/validate"
}

Configs:

{
    "name": "first-opensearch-connector",
    "config": {
        "connector.class": "com.dmathieu.kafka.opensearch.OpenSearchSinkConnector",
        "connection.url": "http://here_is_my_ip:9200",
        "tasks.max": "1",
        "topics": "filebeatData",
        "key.ignore": "true",
        "name": "first-opensearch-connector",
        "connection.username": "admin",
        "connection.password": "admin",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "schema.ignore": "true",
        "type.name": "_doc"
    }
}

I have added connection.username and connection.password with valid values but still getting this error.