[Bug]: Bootstrap endpoint does not return proper dns of the brokers

[Sakib37]

Bug Description

I have a kafka cluster(version: 3.3.2) and kafkaConnect deployed via strimzi operator (v0.33) and it is working fine within same cloud provider(GCP). However, I want to access it from AWS side, so I exposed with externally with N(no. of broker) + 1 internal loadbalancers as below

listeners:
      - name: plain
        port: 9092
        tls: false
        type: internal
      - authentication:
          type: tls
        name: tls
        port: 9093
        tls: true
        type: internal
      - configuration:
          bootstrap:
            alternativeNames:
              - kafka-external-bootstrap.mydomain.com.
            annotations:
              networking.gke.io/internal-load-balancer-allow-global-access: 'true'
              networking.gke.io/load-balancer-type: Internal
            loadBalancerIP: 10.0.0.1
          brokers:
            - advertisedHost: kafka-broker-0.mydomain.com.
              advertisedPort: 9094
              annotations:
                networking.gke.io/load-balancer-type: Internal
              broker: 0
              loadBalancerIP: 10.0.0.2
            - advertisedHost: kafka-broker-1.mydomain.com.
              advertisedPort: 9094
              annotations:
                networking.gke.io/load-balancer-type: Internal
              broker: 1
              loadBalancerIP: 10.0.0.3
            - advertisedHost: kafka-broker-1.mydomain.com.
              advertisedPort: 9094
              annotations:
                networking.gke.io/load-balancer-type: Internal
              broker: 2
              loadBalancerIP: 10.0.0.4
          externalTrafficPolicy: Local
        name: external
        port: 9094
        tls: false
        type: loadbalancer

All the broker dns endpoint are reachable from AWS side on port 9094. However, when I deploy a KafkaConnect in AWS with that external bootstrapServers endpoint, it seems like kafka connect is trying to connect with the internal DNS name of the GKE node where the broker is located instead of the broker dns endpoint(ex: kafka-broker-1.mydomain.com. ) configured in the above configuration. So I get the following error

2023-03-23 14:00:45,838 WARN [AdminClient clientId=adminclient-1] Error connecting to node gke-kafka0node-dns.internal:32422 (id: 1 rack: europe-west1-c) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
java.net.UnknownHostException: gke-kafka0node-dns.internal: Name or service not known
	at java.base/java.net.Inet4AddressImpl.lookupAllHostAddr(Native Method)
	at java.base/java.net.InetAddress$PlatformNameService.lookupAllHostAddr(InetAddress.java:933)
	at java.base/java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1534)
	at java.base/java.net.InetAddress$NameServiceAddresses.get(InetAddress.java:852)
	at java.base/java.net.InetAddress.getAllByName0(InetAddress.java:1524)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1381)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1305)
	at org.apache.kafka.clients.DefaultHostResolver.resolve(DefaultHostResolver.java:27)
	at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:110)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:510)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:467)
	at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:173)
	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:990)
	at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:301)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:1143)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1403)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1346)
	at java.base/java.lang.Thread.run(Thread.java:833)

Steps to reproduce

1. Expose a Kafka cluster externally via loadbalancer and dns using the above configuration.
2. Create a new KafkaConnect outside of the Cloud provider where kafka is deployed and use the external bootstrap endpoint in the KafkaConnect.
3. All the connect pod will fail because the bootstrap endpoint returns the internal dns of the host where kafka broker pod is located instead of the dns configured for the broker. As the host dns is internal to the cloud, Kafkaconnect will fail to connect with the broker.

Expected behavior

The bootstrap endpoint configured in Kafka object, will return the configured dns names of the broker in the configuration instead of the host name of the node where broker pod is located.

Strimzi version

0.33.0

Kubernetes version

GKE, EKS

Installation method

Helm chart

Infrastructure

GCP, AWS

Configuration files and logs

No response

Additional context

No response

[scholzj]

I guess you should explain how did you actually exposed the internal loadbalancers and how you configure the Connect cluster. For the Connect cluster, full log and full custom resources would be definitely useful.

[scholzj]

I mean, internal load balancers are internal => so how do you expect them to be accessed from AWS?

[Sakib37]

As I have mentioned, All the broker dns endpoint are reachable from AWS side on port 9094. However, forgot to mentioned that they are connected via VPN to the GCP cloud. So internal IP routing is working fine. The only problem is KafkaConnect object tries to connect with the dns name of the GCP compute node (which is not resolvable outside of GCP) but it should actually try to connect to the dns name of the broker. FYI, all the broker DNS are reachable on port 9094 from AWS without any issue.
I will post the complete KafkaConnect object soon.

[scholzj]

It is not really clear what does it mean to have a DNS reachable. The fact that the DNS record is known in one location does not mean it can connect to where it resolves to.

[Sakib37]

Please excuse me for the confusion. What I meant by dns reachable is both name and networking is correctly working.

'nc -vz kafka-broker-1.mydomain.com 9094'
OR
'telnet kafka-broker-1.mydomain.com 9094' is successful.

[scholzj]

Well, then I guess you have to share the full logs and configurations as your connect seems to be connecting to gke-kafka0node-dns.internal:32422

[Sakib37]

Here is the template for both kafka(in GCP) and kafkaConnect (in AWS)

Kafka:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-kafka
  namespace: kafka
spec:
  kafka:
    config:
      auto.create.topics.enable: 'true'
      default.replication.factor: 3
      inter.broker.protocol.version: '3.3'
      min.insync.replicas: 2
      offsets.topic.replication.factor: 3
      transaction.state.log.min.isr: 2
      transaction.state.log.replication.factor: 3
    listeners:
      - configuration:
          useServiceDnsDomain: true
        name: plain
        port: 9092
        tls: false
        type: internal
      - authentication:
          type: tls
        name: tls
        port: 9093
        tls: true
        type: internal
      - configuration:
          bootstrap:
            alternativeNames:
              - kafka-external-bootstrap.mydomain.com
            annotations:
              networking.gke.io/internal-load-balancer-allow-global-access: 'true'
              networking.gke.io/load-balancer-type: Internal
            loadBalancerIP: 10.132.0.41
          brokers:
            - advertisedHost: kafka-broker-0.mydomain.com.
              advertisedPort: 9094
              annotations:
                networking.gke.io/internal-load-balancer-allow-global-access: 'true'
                networking.gke.io/load-balancer-type: Internal
              broker: 0
              loadBalancerIP: 10.132.0.68
            - advertisedHost: kafka-broker-1.mydomain.com.
              advertisedPort: 9094
              annotations:
                networking.gke.io/internal-load-balancer-allow-global-access: 'true'
                networking.gke.io/load-balancer-type: Internal
              broker: 1
              loadBalancerIP: 10.132.0.42
            - advertisedHost: kafka-broker-2.mydomain.com.
              advertisedPort: 9094
              annotations:
                networking.gke.io/internal-load-balancer-allow-global-access: 'true'
                networking.gke.io/load-balancer-type: Internal
              broker: 2
              loadBalancerIP: 10.132.0.17
          externalTrafficPolicy: Local
        name: external
        port: 9094
        tls: false
        type: loadbalancer
    logging:
      loggers:
        kafka.root.logger.level: INFO
      type: inline
    readinessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    replicas: 3
    resources:
        ...
    storage:
      type: jbod
      volumes:
        - class: standard
          deleteClaim: false
          id: 0
          size: 50Gi
          type: persistent-claim
    template:
      pod:
        ...
        ...
    version: 3.3.2

KafkaConnect (in AWS):

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-kafka-connect
  namespace: kafka
spec:
  bootstrapServers: kafka-external-bootstrap.mydomain.com.:9094
  config:
    config.storage.replication.factor: -1
    config.storage.topic: my-kafka-connect-connect-cluster-configs
    group.id: my-kafka-connect-connect-cluster
    key.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    offset.storage.replication.factor: -1
    offset.storage.topic: my-kafka-connect-connect-cluster-offsets
    status.storage.replication.factor: -1
    status.storage.topic: my-kafka-connect-connect-cluster-status
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true
  livenessProbe:
    initialDelaySeconds: 300
    timeoutSeconds: 60
  logging:
    loggers:
      log4j.rootLogger: INFO
    type: inline
  replicas: 3
  ....
  ....
  version: 3.3.2

However, it is worth to mention that a similar kafkaConnect in GCP is working fine with similar config. This is because the GCP KafkaConnect can resolve the internal DNS name of the GKE node.

Logs from Kafkaconenct(AWS):

Preparing truststore
Preparing truststore is complete
Starting Kafka Connect with configuration:
# Bootstrap servers
bootstrap.servers=kafka-external-bootstrap.mydomain.com.:9094
# REST Listeners
rest.port=8083
rest.advertised.host.name=172.19.64.62
rest.advertised.port=8083
# Plugins
plugin.path=/opt/kafka/plugins
# Provided configuration
offset.storage.topic=my-kafka-connect-connect-cluster-offsets
value.converter=org.apache.kafka.connect.json.JsonConverter
config.storage.topic=my-kafka-connect-connect-cluster-configs
key.converter=org.apache.kafka.connect.json.JsonConverter
group.id=my-kafka-connect-connect-cluster
status.storage.topic=my-kafka-connect-connect-cluster-status
config.storage.replication.factor=-1
key.converter.schemas.enable=true
offset.storage.replication.factor=-1
status.storage.replication.factor=-1
value.converter.schemas.enable=true


security.protocol=PLAINTEXT
producer.security.protocol=PLAINTEXT
consumer.security.protocol=PLAINTEXT
admin.security.protocol=PLAINTEXT



# Additional configuration
consumer.client.rack=us-east-2c

Configuring Java heap: -Xms5690831667 -Xmx5690831667
+ exec /usr/bin/tini -w -e 143 -- /opt/kafka/bin/connect-distributed.sh /tmp/strimzi-connect.properties
2023-03-24 10:01:57,312 INFO WorkerInfo values:
	jvm.args = -Xms5690831667, -Xmx5690831667, -XX:+UseG1GC, -XX:MaxGCPauseMillis=20, -XX:InitiatingHeapOccupancyPercent=35, -XX:+ExplicitGCInvokesConcurrent, -XX:MaxInlineLevel=15, -Djava.awt.headless=true, -Dcom.sun.management.jmxremote, -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, -javaagent:/opt/kafka/libs/jmx_prometheus_javaagent-0.17.2.jar=9404:/opt/kafka/custom-config/metrics-config.json
	jvm.spec = Red Hat, Inc., OpenJDK 64-Bit Server VM, 17.0.6, 17.0.6+10-LTS
	jvm.classpath = /opt/kafka/bin/../libs/accessors-smart-2.4.7.jar:/opt/kafka/bin/../libs/activation-1.1.1.jar:/opt/kafka/bin/../libs/annotations-13.0.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.5.0.jar:/opt/kafka/bin/../libs/checker-qual-3.5.0.jar:/opt/kafka/bin/../libs/commons-cli-1.4.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-lang3-3.8.1.jar:/opt/kafka/bin/../libs/connect-api-3.3.2.jar:/opt/kafka/bin/../libs/connect-basic-auth-extension-3.3.2.jar:/opt/kafka/bin/../libs/connect-json-3.3.2.jar:/opt/kafka/bin/../libs/connect-mirror-3.3.2.jar:/opt/kafka/bin/../libs/connect-mirror-client-3.3.2.jar:/opt/kafka/bin/../libs/connect-runtime-3.3.2.jar:/opt/kafka/bin/../libs/connect-transforms-3.3.2.jar:/opt/kafka/bin/../libs/cruise-control-metrics-reporter-2.5.111.jar:/opt/kafka/bin/../libs/error_prone_annotations-2.3.4.jar:/opt/kafka/bin/../libs/failureaccess-1.0.1.jar:/opt/kafka/bin/../libs/gson-2.9.0.jar:/opt/kafka/bin/../libs/guava-30.1-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-1.3.jar:/opt/kafka/bin/../libs/jackson-annotations-2.13.4.jar:/opt/kafka/bin/../libs/jackson-core-2.13.4.jar:/opt/kafka/bin/../libs/jackson-databind-2.13.4.2.jar:/opt/kafka/bin/../libs/jackson-dataformat-csv-2.13.4.jar:/opt/kafka/bin/../libs/jackson-dataformat-yaml-2.13.4.jar:/opt/kafka/bin/../libs/jackson-datatype-jdk8-2.13.4.jar:/opt/kafka/bin/../libs/jackson-datatype-jsr310-2.13.4.jar:/opt/kafka/bin/../libs/jackson-jaxrs-base-2.13.4.jar:/opt/kafka/bin/../libs/jackson-jaxrs-json-provider-2.13.4.jar:/opt/kafka/bin/../libs/jackson-module-jaxb-annotations-2.13.4.jar:/opt/kafka/bin/../libs/jackson-module-scala_2.13-2.13.4.jar:/opt/kafka/bin/../libs/jaeger-client-1.8.1.jar:/opt/kafka/bin/../libs/jaeger-core-1.8.1.jar:/opt/kafka/bin/../libs/jaeger-thrift-1.8.1.jar:/opt/kafka/bin/../libs/jaeger-tracerresolver-1.8.1.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.27.0-GA.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.0.jar:/opt/kafka/bin/../libs/jersey-client-2.34.jar:/opt/kafka/bin/../libs/jersey-common-2.34.jar:/opt/kafka/bin/../libs/jersey-container-servlet-2.34.jar:/opt/kafka/bin/../libs/jersey-container-servlet-core-2.34.jar:/opt/kafka/bin/../libs/jersey-hk2-2.34.jar:/opt/kafka/bin/../libs/jersey-server-2.34.jar:/opt/kafka/bin/../libs/jetty-client-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-continuation-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-http-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-io-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-security-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-server-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-servlet-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-servlets-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-util-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jetty-util-ajax-9.4.48.v20220622.jar:/opt/kafka/bin/../libs/jline-3.21.0.jar:/opt/kafka/bin/../libs/jmx_prometheus_javaagent-0.17.2.jar:/opt/kafka/bin/../libs/jopt-simple-5.0.4.jar:/opt/kafka/bin/../libs/jose4j-0.7.9.jar:/opt/kafka/bin/../libs/json-path-2.6.0.jar:/opt/kafka/bin/../libs/json-smart-2.4.7.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.33.0.jar:/opt/kafka/bin/../libs/kafka-clients-3.3.2.jar:/opt/kafka/bin/../libs/kafka-env-var-config-provider-1.1.0.jar:/opt/kafka/bin/../libs/kafka-kubernetes-config-provider-1.1.0.jar:/opt/kafka/bin/../libs/kafka-log4j-appender-3.3.2.jar:/opt/kafka/bin/../libs/kafka-metadata-3.3.2.jar:/opt/kafka/bin/../libs/kafka-oauth-client-0.11.0.jar:/opt/kafka/bin/../libs/kafka-oauth-common-0.11.0.jar:/opt/kafka/bin/../libs/kafka-oauth-keycloak-authorizer-0.11.0.jar:/opt/kafka/bin/../libs/kafka-oauth-server-0.11.0.jar:/opt/kafka/bin/../libs/kafka-oauth-server-plain-0.11.0.jar:/opt/kafka/bin/../libs/kafka-quotas-plugin-0.2.0.jar:/opt/kafka/bin/../libs/kafka-raft-3.3.2.jar:/opt/kafka/bin/../libs/kafka-server-common-3.3.2.jar:/opt/kafka/bin/../libs/kafka-shell-3.3.2.jar:/opt/kafka/bin/../libs/kafka-storage-3.3.2.jar:/opt/kafka/bin/../libs/kafka-storage-api-3.3.2.jar:/opt/kafka/bin/../libs/kafka-streams-3.3.2.jar:/opt/kafka/bin/../libs/kafka-streams-examples-3.3.2.jar:/opt/kafka/bin/../libs/kafka-streams-scala_2.13-3.3.2.jar:/opt/kafka/bin/../libs/kafka-streams-test-utils-3.3.2.jar:/opt/kafka/bin/../libs/kafka-tools-3.3.2.jar:/opt/kafka/bin/../libs/kafka_2.13-3.3.2.jar:/opt/kafka/bin/../libs/kotlin-stdlib-1.4.10.jar:/opt/kafka/bin/../libs/kotlin-stdlib-common-1.4.0.jar:/opt/kafka/bin/../libs/kubernetes-client-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-client-api-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-httpclient-okhttp-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-apps-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-batch-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-certificates-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-common-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-core-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-extensions-6.2.0.jar:/opt/kafka/bin/../libs/kubernetes-model-gatewayapi-6.2.0.jar:/opt/kafka/bin/../libs/libthrift-0.15.0.jar:/opt/kafka/bin/../libs/listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar:/opt/kafka/bin/../libs/logging-interceptor-3.12.12.jar:/opt/kafka/bin/../libs/lz4-java-1.8.0.jar:/opt/kafka/bin/../libs/maven-artifact-3.8.4.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.33.0.jar:/opt/kafka/bin/../libs/netty-buffer-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-codec-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-common-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-handler-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-resolver-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-transport-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-transport-classes-epoll-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-transport-native-epoll-4.1.78.Final.jar:/opt/kafka/bin/../libs/netty-transport-native-unix-common-4.1.78.Final.jar:/opt/kafka/bin/../libs/nimbus-jose-jwt-9.10.jar:/opt/kafka/bin/../libs/okhttp-4.9.3.jar:/opt/kafka/bin/../libs/okio-2.8.0.jar:/opt/kafka/bin/../libs/opa-authorizer-1.5.0.jar:/opt/kafka/bin/../libs/opentelemetry-api-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-context-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-common-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-otlp-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-exporter-otlp-common-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-instrumentation-api-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-instrumentation-api-semconv-1.18.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-kafka-clients-2.6-1.18.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-kafka-clients-common-1.18.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-common-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-extension-autoconfigure-1.18.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-extension-autoconfigure-spi-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-logs-1.18.0-alpha.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-metrics-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-sdk-trace-1.18.0.jar:/opt/kafka/bin/../libs/opentelemetry-semconv-1.18.0-alpha.jar:/opt/kafka/bin/../libs/opentracing-api-0.33.0.jar:/opt/kafka/bin/../libs/opentracing-kafka-client-0.1.15.jar:/opt/kafka/bin/../libs/opentracing-noop-0.33.0.jar:/opt/kafka/bin/../libs/opentracing-tracerresolver-0.1.8.jar:/opt/kafka/bin/../libs/opentracing-util-0.33.0.jar:/opt/kafka/bin/../libs/osgi-resource-locator-1.0.3.jar:/opt/kafka/bin/../libs/paranamer-2.8.jar:/opt/kafka/bin/../libs/plexus-utils-3.3.0.jar:/opt/kafka/bin/../libs/reflections-0.9.12.jar:/opt/kafka/bin/../libs/reload4j-1.2.19.jar:/opt/kafka/bin/../libs/rocksdbjni-7.1.2.jar:/opt/kafka/bin/../libs/scala-collection-compat_2.13-2.6.0.jar:/opt/kafka/bin/../libs/scala-java8-compat_2.13-1.0.2.jar:/opt/kafka/bin/../libs/scala-library-2.13.8.jar:/opt/kafka/bin/../libs/scala-logging_2.13-3.9.4.jar:/opt/kafka/bin/../libs/scala-reflect-2.13.8.jar:/opt/kafka/bin/../libs/slf4j-api-1.7.30.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-1.33.jar:/opt/kafka/bin/../libs/snappy-java-1.1.8.4.jar:/opt/kafka/bin/../libs/swagger-annotations-2.2.0.jar:/opt/kafka/bin/../libs/tracing-agent-0.33.0.jar:/opt/kafka/bin/../libs/trogdor-3.3.2.jar:/opt/kafka/bin/../libs/zjsonpatch-0.3.0.jar:/opt/kafka/bin/../libs/zookeeper-3.6.3.jar:/opt/kafka/bin/../libs/zookeeper-jute-3.6.3.jar:/opt/kafka/bin/../libs/zstd-jni-1.5.2-1.jar
	os.spec = Linux, amd64, 5.4.226-129.415.amzn2.x86_64
	os.vcpus = 2
 (org.apache.kafka.connect.runtime.WorkerInfo) [main]
2023-03-24 10:01:57,321 INFO Scanning for plugin classes. This might take a moment ... (org.apache.kafka.connect.cli.ConnectDistributed) [main]
2023-03-24 10:02:01,778 INFO Registered loader: jdk.internal.loader.ClassLoaders$AppClassLoader@7ad041f3 (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,778 INFO Added plugin 'org.apache.kafka.connect.tools.VerifiableSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,778 INFO Added plugin 'org.apache.kafka.connect.tools.MockSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,779 INFO Added plugin 'org.apache.kafka.connect.tools.VerifiableSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,779 INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,779 INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,779 INFO Added plugin 'org.apache.kafka.connect.tools.SchemaSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,779 INFO Added plugin 'org.apache.kafka.connect.tools.MockSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,780 INFO Added plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,780 INFO Added plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,780 INFO Added plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,781 INFO Added plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,781 INFO Added plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,781 INFO Added plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,781 INFO Added plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,782 INFO Added plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,782 INFO Added plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,782 INFO Added plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,783 INFO Added plugin 'org.apache.kafka.connect.transforms.ReplaceField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,783 INFO Added plugin 'org.apache.kafka.connect.transforms.SetSchemaMetadata$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,783 INFO Added plugin 'org.apache.kafka.connect.transforms.ReplaceField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,783 INFO Added plugin 'org.apache.kafka.connect.transforms.Filter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,783 INFO Added plugin 'org.apache.kafka.connect.transforms.InsertField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,783 INFO Added plugin 'org.apache.kafka.connect.transforms.HeaderFrom$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,784 INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampConverter$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,784 INFO Added plugin 'org.apache.kafka.connect.transforms.MaskField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,784 INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,784 INFO Added plugin 'org.apache.kafka.connect.transforms.RegexRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,785 INFO Added plugin 'org.apache.kafka.connect.transforms.HoistField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,785 INFO Added plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,785 INFO Added plugin 'org.apache.kafka.connect.transforms.MaskField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,785 INFO Added plugin 'org.apache.kafka.connect.transforms.DropHeaders' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,785 INFO Added plugin 'org.apache.kafka.connect.transforms.Cast$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,785 INFO Added plugin 'org.apache.kafka.connect.transforms.Cast$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,786 INFO Added plugin 'org.apache.kafka.connect.runtime.PredicatedTransformation' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,786 INFO Added plugin 'org.apache.kafka.connect.transforms.ExtractField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,786 INFO Added plugin 'org.apache.kafka.connect.transforms.Flatten$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,786 INFO Added plugin 'org.apache.kafka.connect.transforms.InsertHeader' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,786 INFO Added plugin 'org.apache.kafka.connect.transforms.InsertField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,786 INFO Added plugin 'org.apache.kafka.connect.transforms.Flatten$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,787 INFO Added plugin 'org.apache.kafka.connect.transforms.HeaderFrom$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,788 INFO Added plugin 'org.apache.kafka.connect.transforms.SetSchemaMetadata$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,788 INFO Added plugin 'org.apache.kafka.connect.transforms.ExtractField$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,789 INFO Added plugin 'org.apache.kafka.connect.transforms.TimestampConverter$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,789 INFO Added plugin 'org.apache.kafka.connect.transforms.HoistField$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,789 INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.HasHeaderKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,789 INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.RecordIsTombstone' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,789 INFO Added plugin 'org.apache.kafka.connect.transforms.predicates.TopicNameMatches' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,789 INFO Added plugin 'org.apache.kafka.common.config.provider.FileConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,790 INFO Added plugin 'org.apache.kafka.common.config.provider.DirectoryConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,790 INFO Added plugin 'io.strimzi.kafka.EnvVarConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,790 INFO Added plugin 'io.strimzi.kafka.KubernetesConfigMapConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,790 INFO Added plugin 'io.strimzi.kafka.KubernetesSecretConfigProvider' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,791 INFO Added plugin 'org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,791 INFO Added plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,791 INFO Added plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,791 INFO Added plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,795 INFO Added aliases 'MirrorCheckpointConnector' and 'MirrorCheckpoint' to plugin 'org.apache.kafka.connect.mirror.MirrorCheckpointConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,795 INFO Added aliases 'MirrorHeartbeatConnector' and 'MirrorHeartbeat' to plugin 'org.apache.kafka.connect.mirror.MirrorHeartbeatConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,796 INFO Added aliases 'MirrorSourceConnector' and 'MirrorSource' to plugin 'org.apache.kafka.connect.mirror.MirrorSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,796 INFO Added aliases 'MockSinkConnector' and 'MockSink' to plugin 'org.apache.kafka.connect.tools.MockSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,797 INFO Added aliases 'MockSourceConnector' and 'MockSource' to plugin 'org.apache.kafka.connect.tools.MockSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,797 INFO Added aliases 'SchemaSourceConnector' and 'SchemaSource' to plugin 'org.apache.kafka.connect.tools.SchemaSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,798 INFO Added aliases 'VerifiableSinkConnector' and 'VerifiableSink' to plugin 'org.apache.kafka.connect.tools.VerifiableSinkConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,798 INFO Added aliases 'VerifiableSourceConnector' and 'VerifiableSource' to plugin 'org.apache.kafka.connect.tools.VerifiableSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,799 INFO Added aliases 'ByteArrayConverter' and 'ByteArray' to plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,799 INFO Added aliases 'DoubleConverter' and 'Double' to plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,799 INFO Added aliases 'FloatConverter' and 'Float' to plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,799 INFO Added aliases 'IntegerConverter' and 'Integer' to plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,800 INFO Added aliases 'LongConverter' and 'Long' to plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,800 INFO Added aliases 'ShortConverter' and 'Short' to plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,800 INFO Added aliases 'JsonConverter' and 'Json' to plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,801 INFO Added aliases 'StringConverter' and 'String' to plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,801 INFO Added aliases 'ByteArrayConverter' and 'ByteArray' to plugin 'org.apache.kafka.connect.converters.ByteArrayConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,801 INFO Added aliases 'DoubleConverter' and 'Double' to plugin 'org.apache.kafka.connect.converters.DoubleConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,801 INFO Added aliases 'FloatConverter' and 'Float' to plugin 'org.apache.kafka.connect.converters.FloatConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,802 INFO Added aliases 'IntegerConverter' and 'Integer' to plugin 'org.apache.kafka.connect.converters.IntegerConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,802 INFO Added aliases 'LongConverter' and 'Long' to plugin 'org.apache.kafka.connect.converters.LongConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,802 INFO Added aliases 'ShortConverter' and 'Short' to plugin 'org.apache.kafka.connect.converters.ShortConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,802 INFO Added aliases 'JsonConverter' and 'Json' to plugin 'org.apache.kafka.connect.json.JsonConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,802 INFO Added aliases 'SimpleHeaderConverter' and 'Simple' to plugin 'org.apache.kafka.connect.storage.SimpleHeaderConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,803 INFO Added aliases 'StringConverter' and 'String' to plugin 'org.apache.kafka.connect.storage.StringConverter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,803 INFO Added aliases 'PredicatedTransformation' and 'Predicated' to plugin 'org.apache.kafka.connect.runtime.PredicatedTransformation' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,804 INFO Added alias 'DropHeaders' to plugin 'org.apache.kafka.connect.transforms.DropHeaders' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,805 INFO Added alias 'Filter' to plugin 'org.apache.kafka.connect.transforms.Filter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,805 INFO Added alias 'InsertHeader' to plugin 'org.apache.kafka.connect.transforms.InsertHeader' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,805 INFO Added alias 'RegexRouter' to plugin 'org.apache.kafka.connect.transforms.RegexRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,806 INFO Added alias 'TimestampRouter' to plugin 'org.apache.kafka.connect.transforms.TimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,806 INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,807 INFO Added alias 'HasHeaderKey' to plugin 'org.apache.kafka.connect.transforms.predicates.HasHeaderKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,807 INFO Added alias 'RecordIsTombstone' to plugin 'org.apache.kafka.connect.transforms.predicates.RecordIsTombstone' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,808 INFO Added alias 'TopicNameMatches' to plugin 'org.apache.kafka.connect.transforms.predicates.TopicNameMatches' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,809 INFO Added alias 'BasicAuthSecurityRestExtension' to plugin 'org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,809 INFO Added aliases 'AllConnectorClientConfigOverridePolicy' and 'All' to plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,809 INFO Added aliases 'NoneConnectorClientConfigOverridePolicy' and 'None' to plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,809 INFO Added aliases 'PrincipalConnectorClientConfigOverridePolicy' and 'Principal' to plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-03-24 10:02:01,925 INFO DistributedConfig values:
	access.control.allow.methods =
	access.control.allow.origin =
	admin.listeners = null
	bootstrap.servers = [kafka-external-bootstrap.mydomain.com.:9094]
	client.dns.lookup = use_all_dns_ips
	client.id =
	config.providers = []
	config.storage.replication.factor = -1
	config.storage.topic = my-kafka-connect-connect-cluster-configs
	connect.protocol = sessioned
	connections.max.idle.ms = 540000
	connector.client.config.override.policy = All
	exactly.once.source.support = disabled
	group.id = my-kafka-connect-connect-cluster
	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]
	key.converter = class org.apache.kafka.connect.json.JsonConverter
	listeners = [http://:8083]
	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 = -1
	offset.storage.topic = my-kafka-connect-connect-cluster-offsets
	plugin.path = [/opt/kafka/plugins]
	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 = 172.19.64.62
	rest.advertised.listener = null
	rest.advertised.port = 8083
	rest.extension.classes = []
	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.connect.timeout.ms = null
	sasl.login.read.timeout.ms = 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.login.retry.backoff.max.ms = 10000
	sasl.login.retry.backoff.ms = 100
	sasl.mechanism = GSSAPI
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = sub
	sasl.oauthbearer.token.endpoint.url = null
	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 = -1
	status.storage.topic = my-kafka-connect-connect-cluster-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) [main]
2023-03-24 10:02:01,928 INFO Creating Kafka admin client (org.apache.kafka.connect.util.ConnectUtils) [main]
2023-03-24 10:02:01,940 INFO AdminClientConfig values:
	bootstrap.servers = [kafka-external-bootstrap.mydomain.com.:9094]
	client.dns.lookup = use_all_dns_ips
	client.id =
	connections.max.idle.ms = 300000
	default.api.timeout.ms = 60000
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	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.connect.timeout.ms = null
	sasl.login.read.timeout.ms = 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.login.retry.backoff.max.ms = 10000
	sasl.login.retry.backoff.ms = 100
	sasl.mechanism = GSSAPI
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = sub
	sasl.oauthbearer.token.endpoint.url = null
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	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
 (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2023-03-24 10:02:02,224 WARN These configurations '[consumer.client.rack, config.storage.topic, producer.security.protocol, rest.advertised.host.name, group.id, status.storage.topic, rest.advertised.port, plugin.path, admin.security.protocol, rest.port, config.storage.replication.factor, key.converter.schemas.enable, status.storage.replication.factor, value.converter.schemas.enable, offset.storage.replication.factor, offset.storage.topic, consumer.security.protocol, value.converter, key.converter]' were supplied but are not used yet. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2023-03-24 10:02:02,225 INFO Kafka version: 3.3.2 (org.apache.kafka.common.utils.AppInfoParser) [main]
2023-03-24 10:02:02,225 INFO Kafka commitId: b66af662e61082cb (org.apache.kafka.common.utils.AppInfoParser) [main]
2023-03-24 10:02:02,225 INFO Kafka startTimeMs: 1679652122224 (org.apache.kafka.common.utils.AppInfoParser) [main]
2023-03-24 10:02:03,050 WARN [AdminClient clientId=adminclient-1] Error connecting to node gke-my-kafka-01-kafka-a369c241-v7wn.c.dh-darkstores-stg.internal:31095 (id: 0 rack: europe-west1-b) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
java.net.UnknownHostException: gke-my-kafka-01-kafka-a369c241-v7wn.c.dh-darkstores-stg.internal: Name or service not known
	at java.base/java.net.Inet4AddressImpl.lookupAllHostAddr(Native Method)
	at java.base/java.net.InetAddress$PlatformNameService.lookupAllHostAddr(InetAddress.java:933)
	at java.base/java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1534)
	at java.base/java.net.InetAddress$NameServiceAddresses.get(InetAddress.java:852)
	at java.base/java.net.InetAddress.getAllByName0(InetAddress.java:1524)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1381)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1305)
	at org.apache.kafka.clients.DefaultHostResolver.resolve(DefaultHostResolver.java:27)
	at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:110)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:510)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:467)
	at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:173)
	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:990)
	at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:301)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:1143)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1403)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1346)
	at java.base/java.lang.Thread.run(Thread.java:833)
2023-03-24 10:02:03,115 WARN [AdminClient clientId=adminclient-1] Error connecting to node gke-my-kafka-01-kafka-db995e63-cdfc.c.dh-darkstores-stg.internal:32422 (id: 1 rack: europe-west1-c) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
java.net.UnknownHostException: gke-my-kafka-01-kafka-db995e63-cdfc.c.dh-darkstores-stg.internal: Name or service not known
	at java.base/java.net.Inet4AddressImpl.lookupAllHostAddr(Native Method)
	at java.base/java.net.InetAddress$PlatformNameService.lookupAllHostAddr(InetAddress.java:933)
	at java.base/java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1534)
	at java.base/java.net.InetAddress$NameServiceAddresses.get(InetAddress.java:852)
	at java.base/java.net.InetAddress.getAllByName0(InetAddress.java:1524)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1381)
	at java.base/java.net.InetAddress.getAllByName(InetAddress.java:1305)
	at org.apache.kafka.clients.DefaultHostResolver.resolve(DefaultHostResolver.java:27)
	at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:110)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:510)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:467)
	at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:173)
	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:990)
	at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:301)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:1143)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1403)
	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1346)
	at java.base/java.lang.Thread.run(Thread.java:833)

[scholzj]

The . at the end of the bootstrap address should normally not be there: kafka-external-bootstrap.mydomain.com. => but not really sure if that matters. But you should in any case try to remove it and see if something changes.

---

Can you also share the Kafka CR => complete including the status as you get it with kubectl get kafka -o yaml?

[Sakib37]

Hi @scholzj
The . at the end of the bootstrap address should normally not be there
Seems like this is the actual issue. I removed the . from the end of the domain name and the error is gone. Thanks a lot for such a quick help.
However, I am still wondering if there is any bootstrap api endpoint (ex- http://bootstrap-endpoint:9094/brokers) where I can verify what broker list is returned by the bootstrap endpoint.

[scholzj]

Seems like this is the actual issue. I removed the . from the end of the domain name and the error is gone. Thanks a lot for such a quick help.

Interesting ... I did not really expect this to be the issue. But glad it helped :-D

However, I am still wondering if there is any bootstrap api endpoint (ex- http://bootstrap-endpoint:9094/brokers) where I can verify what broker list is returned by the bootstrap endpoint.

No, Kafka does not have any endpoint like that. You should see the bootstrap addresses recommended by the operator in the .status section of the Kafka custom resource.