Kafka Connect消费者连接Redpanda Broker失败求助

报错信息 #

bootstrap broker backbone.redpanda.svc.cluster.local:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

当前部署配置 #

Kafka Connect Deployment #

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-connect
  namespace: s3-connector
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-connect
  template:
    metadata:
      labels:
        app: kafka-connect
    spec:
      containers:
        - name: kafka-connect
          image: confluentinc/cp-kafka-connect:6.1.9
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8083
          securityContext:
            runAsUser: 0
          env:
            - name: CONNECT_BOOTSTRAP_SERVERS
              value: "backbone.redpanda.svc.cluster.local:9092"
            - name: CONNECT_KAFKA_HEAP_OPTS
              value: "-Xms256M -Xmx2G"
            - name: CONNECT_OFFSETS_TOPIC_REPLICATION_FACTOR
              value: "1"
            - name: CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR
              value: "1"
            - name: CONNECT_STATUS_STORAGE_REPLICATION_FACTOR
              value: "1"
#            - name: CONNECT_LOG4J_ROOT_LOGLEVEL
#              value: "DEBUG"
            - name: CONNECT_REST_PORT
              value: "8083"
            - name: CONNECT_GROUP_ID
              value: "kafka-connect-worker"
            - name: CONNECT_CONFIG_STORAGE_TOPIC
              value: "kafka-connect-configs"
            - name: CONNECT_OFFSET_STORAGE_TOPIC
              value: "kafka-connect-offsets"
            - name: CONNECT_STATUS_STORAGE_TOPIC
              value: "kafka-connect-statuses"
            - name: CONNECT_REST_ADVERTISED_HOST_NAME
              value: "kafka-connect:8083"
            - name: CONNECT_KEY_CONVERTER
              value: "org.apache.kafka.connect.json.JsonConverter"
            - name: CONNECT_VALUE_CONVERTER
              value: "org.apache.kafka.connect.json.JsonConverter"
            - name: CONNECT_PLUGIN_PATH
              value: "/usr/share/java,/usr/share/confluent-hub-components,/data/connect-jars"
            - name: AWS_ACCESS_KEY_ID
              value: "---"
            - name: AWS_SECRET_ACCESS_KEY
              value: "---"
            - name: CONNECT_LOG4J_LOGGERS
              value: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR"
            - name: CONNECT_SECURITY_PROTOCOL
              value: SSL
            - name: CONNECT_SSL_ENABLED_PROTOCOLS
              value: "TLSv1.2"
            - name: CONNECT_SSL_TRUSTSTORE_LOCATION
              value: "/var/private/ssl/truststore.jks"
            - name: CONNECT_SSL_TRUSTSTORE_PASSWORD
              value: "backbone"
          volumeMounts:
            - name: connect-config-props
              mountPath: /home/appuser
            - name: h2-bundle
              mountPath: /etc/tls-bundle
            - mountPath: /var/private/ssl
              name: tlscert
          command:
            - bash
            - -c
            - |
              echo "Installing Connector"
              confluent-hub install --no-prompt confluentinc/kafka-connect-s3:10.4.0
              confluent-hub install --no-prompt mdrogalis/voluble:0.1.0
              #
              echo "Launching Kafka Connect worker"
              /etc/confluent/docker/run &
              #
              sleep infinity
      volumes:
        - name: connect-config-props
          configMap:
            name: connect-config
        - name: h2-bundle
          configMap:
            name: h2-bundle
        - name: truststore-thing
          configMap:
            name: client-cert
        - name: tlscert
          secret:
            defaultMode: 420
            items:
              - key: truststore.jks
                path: truststore.jks
            secretName: redpanda-client-cert

属性ConfigMap配置 #

apiVersion: v1
kind: ConfigMap
metadata:
  name: connect-config
  namespace: s3-connector
data:
  connect.properties: |-
    bootstrap.servers=backbone.redpanda.svc.cluster.local:9092
    connector.class=io.confluent.connect.s3.S3SinkConnector
    tasks.max=1
    topics=demo
    s3.bucket.name=kafka-connect-s3-demo-devops
    s3.region=eu-west-1
    flush.size=3
    storage.class=io.confluent.connect.s3.storage.S3Storage
    format.class=io.confluent.connect.s3.format.json.JsonFormat
    partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
    schema.compatibility=NONE
    security.protocol=SSL
    ssl.enabled.protocols=TLSv1.2
    ssl.truststore.location=/var/private/ssl/truststore.jks
    ssl.truststore.password=backbone
    consumer.bootstrap.servers=backbone.redpanda.svc.cluster.local:9092
    consumer.security.protocol=SSL
    consumer.ssl.truststore.location=/var/private/ssl/truststore.jks
    consumer.ssl.truststore.password=backbone

测试验证情况 #

• 正常工作的命令：显式指定客户端配置文件时可连接消费

kafka-console-consumer --bootstrap-server backbone.redpanda.svc.cluster.local:9092 --topic demo --from-beginning --consumer.config client.properties

• 失败的命令：未指定配置文件时无法连接

kafka-console-consumer --bootstrap-server backbone.redpanda.svc.cluster.local:9092 --topic demo --from-beginning

• S3 Sink连接器问题：通过API创建连接器时未指定SSL相关配置，同样无法连接Redpanda

curl -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/confluentinc-kafka-connect-s3/config \
    -d '
 {
        "connector.class": "io.confluent.connect.s3.S3SinkConnector",
        "key.converter":"org.apache.kafka.connect.storage.StringConverter",
        "tasks.max": "1",
        "topics": "demo",
        "s3.region": "eu-west-1",
        "s3.bucket.name": "kafka-connect-s3-demo-devops",
        "flush.size": "1",
        "storage.class": "io.confluent.connect.s3.storage.S3Storage",
        "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
        "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
    }
'

问题分析与解决方案 #

核心问题 #

1. 挂载的ConfigMap未被Kafka Connect加载：当前将connect-config ConfigMap挂载到/home/appuser目录，但Kafka Connect默认不会自动加载该路径下的connect.properties文件。
2. 连接器配置缺少SSL参数：通过API创建S3连接器时未指定SSL相关配置，导致连接器无法通过SSL连接Redpanda集群。

具体修复步骤 #

1. 确保Kafka Connect加载自定义配置文件

修改Deployment中的启动命令，指定加载挂载的connect.properties：

# 修改原启动命令部分
echo "Launching Kafka Connect worker"
/etc/confluent/docker/run /home/appuser/connect.properties &

或者通过环境变量指定配置文件路径，在Deployment的env中添加：

- name: CONNECT_CONFIGURATION_FILE
  value: "/home/appuser/connect.properties"

2. 修正连接器配置，添加SSL参数

通过API创建连接器时，必须包含SSL相关配置：

curl -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/confluentinc-kafka-connect-s3/config \
    -d '
 {
        "connector.class": "io.confluent.connect.s3.S3SinkConnector",
        "key.converter":"org.apache.kafka.connect.storage.StringConverter",
        "tasks.max": "1",
        "topics": "demo",
        "s3.region": "eu-west-1",
        "s3.bucket.name": "kafka-connect-s3-demo-devops",
        "flush.size": "1",
        "storage.class": "io.confluent.connect.s3.storage.S3Storage",
        "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
        "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
        "consumer.security.protocol": "SSL",
        "consumer.ssl.enabled.protocols": "TLSv1.2",
        "consumer.ssl.truststore.location": "/var/private/ssl/truststore.jks",
        "consumer.ssl.truststore.password": "backbone"
    }
'

3. 验证配置加载情况

进入Kafka Connect容器，检查配置文件是否存在且可读取：

kubectl exec -n s3-connector -it <kafka-connect-pod-name> -- cat /home/appuser/connect.properties

查看Kafka Connect启动日志，确认配置是否被正确加载：

kubectl logs -n s3-connector <kafka-connect-pod-name>

4. 额外检查点

• 确认redpanda-client-cert Secret中的truststore.jks文件路径和权限正确，容器内/var/private/ssl/truststore.jks可访问。
• 测试网络连通性：在Kafka Connect容器内执行telnet backbone.redpanda.svc.cluster.local 9092，确认端口可访问。

内容的提问来源于stack exchange，提问作者Jyoti Singh