Problem with Strimzi Kafka Connect + Strimzi KafkaConnectors + external Кафка (not Strimzi) using SSL.

[vsfomin]

When running Kafka Connect without SSL, the connectors work fine; however, once SSL is enabled, everything breaks. The error is as follows:
[kafka-producer-network-thread | nexus-wms.data-schemahistory] NetworkClient:1072 - [Producer clientId=nexus-wms.data-schemahistory] Node -3 disconnected. 2025-10-01 16:19:21 INFO [kafka-producer-network-thread | nexus-wms.data-schemahistory] NetworkClient:411 - [Producer clientId=nexus-wms.data-schemahistory] Cancelled in-flight API_VERSIONS request with correlation id 183 due to node -3 being disconnected (elapsed time since creation: 60ms, elapsed time since send: 60ms, throttle time: 0ms, request timeout: 30000ms) 2025-10-01 16:19:21 WARN [kafka-producer-network-thread | nexus-wms.data-schemahistory] NetworkClient:1255 - [Producer clientId=nexus-wms.data-schemahistory] Bootstrap broker ldh-kafka-01-broker-2.ldh-kafka-01-broker-headless:9094 (id: -3 rack: null isFenced: false) disconnected 2025-10-01 16:19:21 INFO [SourceTaskOffsetCommitter-1] BaseSourceTask:473 - Couldn't commit processed log positions with the source database due to a concurrent connector shutdown or restart

Tried connecting to Redpanda — the same error occurs.
If I disable SSL, everything starts working. At the same time, I don’t see any SSL-related errors.
When I deployed Kafka using the Strimzi Operator, it works fine even with SSL enabled.

I use secret ldh-kafka-01-tls was created by Kafka Bitnami installation (self-signed CA) via cert-manager. This is 100% worked certificates, I use it for Kafka connectivity.

This is part of KafkaConnect configuration:

  tls:
     trustedCertificates:
       - secretName: ldh-kafka-01-tls
         certificate: ca.crt
   authentication:
     type: tls
     certificateAndKey:
       secretName: ldh-kafka-01-tls
       certificate: tls.crt
       key: tls.key

[scholzj]

The operator does not know if it is Strimzi-manager Apache Kafka on the other side or not. It configures the TLS always the same way and it works fine with Strimzi. So I would expect this is some configuration error in how you configured it. But without detailed knowledge of how is the other side configured and what it does, and how is the Strimzi side configured, it is hard to say what the issue might be.

The log snippet you shared is badly formatted. But I think nothing inside even suggests there are some TLS issues. TLS issues normally have some handshake errors etc. (That said, Kafka logging can be wierd at times, so who knows). The snippet does not make it even clear where does the error come from. Connect? Some connector?

[vsfomin]

debezium-connect-cluster-wms-01-connect-0_debezium-connect-cluster-wms-01-connect (9).log
Yes, you are right.
I see SSL handshake error in brokers logs:

(channelId=10.233.68.143:9094-10.233.69.222:52494-5-322) (SSL handshake failed) (org.apache.kafka.common.network.Selector)
[2025-10-01 12:09:09,576] INFO [SocketServer listenerType=BROKER, nodeId=100] Failed authentication with /10.233.69.222 (channelId=10.233.68.143:9094-10.233.69.222:52498-3-323) (SSL handshake failed) (org.apache.kafka.common.network.Selector)
[2025-10-01 12:09:10,013] INFO [SocketServer listenerType=BROKER, nodeId=100] Failed authentication with /10.233.69.222 (channelId=10.233.68.143:9094-10.233.69.222:52506-4-323) (SSL handshake failed) (org.apache.kafka.common.network.Selector)
[2025-10-01 12:09:10,208] INFO [SocketServer listenerType=BROKER, nodeId=100] Failed authentication with /10.233.69.222 (channelId=10.233.68.143:9094-10.233.69.222:52508-5-323) (SSL handshake failed) (org.apache.kafka.common.network.Selector)
[2025-10-01 12:09:10,265] INFO [SocketServer listenerType=BROKER, nodeId=100] Failed authentication with /10.233.69.222 

I use mTLS in my case. Is it correct tls and authentification config?
Spec of my KafkaConnect below:

spec:
  authentication:
    certificateAndKey:
      certificate: tls.crt
      key: tls.key
      secretName: ldh-kafka-01-tls
    type: tls
  bootstrapServers: >-
    ldh-kafka-01-broker-0.ldh-kafka-01-broker-headless:9094,ldh-kafka-01-broker-1.ldh-kafka-01-broker-headless:9094,ldh-kafka-01-broker-2.ldh-kafka-01-broker-headless:9094
  config:
    cleanup.policy: compact
    client.id: nexus-debezium
    config.providers: file,secrets,configmap
    config.providers.configmap.class: io.strimzi.kafka.KubernetesConfigMapConfigProvider
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider
    config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
    config.storage.replication.factor: -1
    config.storage.topic: nexus-nexus-debezium.tech.configs
    connector.client.config.override.policy: All
    group.id: nexus-debezium
    key.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    offset.storage.replication.factor: -1
    offset.storage.topic: nexus-nexus-debezium.tech.offsets
    status.storage.replication.factor: -1
    status.storage.topic: nexus-nexus-debezium.tech.status
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true
  image: >-
    docker-dmp-public-prod-local.x5.ru/debezium-strimzi/debezium-strimzi:3cd5bba1
  jmxOptions: {}
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        key: metrics-config.yml
        name: connect-metrics
  replicas: 1
  resources:
    limits:
      cpu: '10'
      memory: 66Gi
    requests:
      cpu: '1'
      memory: 55Gi
  template:
    pod:
      tmpDirSizeLimit: 5Gi
      volumes:
        - name: connector-config
          secret:
            secretName: debezium-secret
  tls:
    trustedCertificates:
      - certificate: ca.crt
        secretName: ldh-kafka-01-tls

advertised.listeners=CLIENT://ldh-kafka-01-broker-0.ldh-kafka-01-broker-headless.ldh-kafka-01.svc.cluster.local:9092,INTERNAL://ldh-kafka-01-broker-0.ldh-kafka-01-broker-headless.ldh-kafka-01.svc.cluster.local:9094,EXTERNAL://ldh-kafka-01.example.com:31196
allow.everyone.if.no.acl.found=true
auto.create.topics.enable=true
controller.listener.names=CONTROLLER
controller.quorum.bootstrap.servers=ldh-kafka-01-controller-0.ldh-kafka-01-controller-headless.ldh-kafka-01.svc.cluster.local:9093,ldh-kafka-01-controller-1.ldh-kafka-01-controller-headless.ldh-kafka-01.svc.cluster.local:9093,ldh-kafka-01-controller-2.ldh-kafka-01-controller-headless.ldh-kafka-01.svc.cluster.local:9093
default.replication.factor=3
inter.broker.listener.name=INTERNAL
listener.security.protocol.map=CONTROLLER:SSL,CLIENT:SSL,INTERNAL:SSL,EXTERNAL:SSL
listeners=CLIENT://:9092,INTERNAL://:9094,EXTERNAL://:9095
log.dir=/bitnami/kafka/data
log.retention.bytes=1073741824
logs.dir=/opt/bitnami/kafka/logs
process.roles=broker
ssl.client.auth=required
ssl.endpoint.identification.algorithm=
ssl.keystore.location=/opt/bitnami/kafka/config/certs/kafka.keystore.jks
ssl.keystore.type=JKS
ssl.truststore.location=/opt/bitnami/kafka/config/certs/kafka.truststore.jks
ssl.truststore.type=JKS
node.id=100

ssl.keystore.password=***
ssl.truststore.password=***

[scholzj]

Well, I'm not sure what version are you running or how do you deploy / configure connectors. The Connect configuration looks fine and working. But some connectors seem to be misconfigured. For example, there are lot of messages about some Oracle connector tasks starting with custom configuration. E.g.:

2025-10-01 16:43:09 INFO  [task-thread-oracle-connector-wms-01-0200-0] AbstractConfig:371 - ProducerConfig values: 
	...
	bootstrap.servers = [ldh-kafka-01-broker-0.ldh-kafka-01-broker-headless:9092, ldh-kafka-01-broker-1.ldh-kafka-01-broker-headless:9092, ldh-kafka-01-broker-2.ldh-kafka-01-broker-headless:9092]
	...
	security.protocol = PLAINTEXT
	...

These seem to be also those failing later I think. So I guess you have some custom configurations there which you need to fix.

[vsfomin]

From what I understand, I need to convert the PEM certificates to JKS format, mount them in the KafkaConnect Pod, and then configure the path to the keystore and truststore, as well as security.protocol: ssl in the spec.config of the KafkaConnector resource. Is this roughly the correct approach, or is there another recommended method?

Something like this:
schema.history.internal.producer.security.protocol=SSL
schema.history.internal.producer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
schema.history.internal.producer.ssl.keystore.password=test1234
schema.history.internal.producer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
schema.history.internal.producer.ssl.truststore.password=test1234
schema.history.internal.producer.ssl.key.password=test1234

schema.history.internal.consumer.security.protocol=SSL
schema.history.internal.consumer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
schema.history.internal.consumer.ssl.keystore.password=test1234
schema.history.internal.consumer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
schema.history.internal.consumer.ssl.truststore.password=test1234
schema.history.internal.consumer.ssl.key.password=test1234

My KafkaConnector config below:

{{- define "mychart.kafkaConnectorSpec.yard-template" -}}
class: io.debezium.connector.oracle.OracleConnector
tasksMax: 1
config:
  database.server.name: connector-yms-msk-dpro-ora581-{{ .connID }}
  database.user: ${secrets:debezium-credentials-{{ .connectionName }}:username}
  database.password: ${secrets:debezium-credentials-{{ .connectionName }}:password}
  database.url: jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=autossh-jumphost-{{ .connectionName }}-service)(PORT=1521))(CONNECT_DATA=(SERVICE_NAME=YMS)))
  database.dbname: YMS
  table.include.list: YMS_PRD.CAR_BRAND, YMS_PRD.CONTRACTOR, YMS_PRD.DELIVERY_TYPE, YMS_PRD.DOC_GATE, YMS_PRD.DRIVER, YMS_PRD.GATE, YMS_PRD.GATE_SETTING, YMS_PRD.INT2YMS, YMS_PRD.INT4YMS, YMS_PRD.LOCATION, YMS_PRD.LOCATION_TYPE, YMS_PRD.NOMENCLATURE, YMS_PRD.NORMATIVE, YMS_PRD.PRIORITY_DC, YMS_PRD.RECEIPT, YMS_PRD.RECEIPT_ALL_REPORT_DATA, YMS_PRD.RECEIPT_REPORT_DATA, YMS_PRD.SCHEDULE_DC, YMS_PRD.SHIPMENT, YMS_PRD.SHIPMENT_ALL_REPORT_DATA, YMS_PRD.SHIPMENT_REPORT_DATA, YMS_PRD.STATUS2YMS, YMS_PRD.STATUS4YMS, YMS_PRD.STOP_LIST, YMS_PRD.TERMINAL_MESSAGE, YMS_PRD.TRANSPORT_TYPE, YMS_PRD.VISIT_DC, YMS_PRD.VISIT_REASON, YMS_PRD.VOICE_CALL_LOG, YMS_PRD.WAREHOUSE, RAF_PRD.STATE_TRANSITION_JOURNAL   
  log.mining.strategy: hybrid
  archive.destination.name: LOG_ARCHIVE_DEST_1
  schema.history.internal.kafka.bootstrap.servers: "{{ .bootstrap_servers }}"
  schema.history.internal.kafka.topic: yms.msk-dpro-ora581.tech.history
  schema.history.internal.store.only.captured.tables.ddl: true
  schema.history.internal.store.only.captured.databases.ddl: true
  snapshot.mode: no_data
  skip.messages.without.change: false
  # tombstones.on.delete: false
  producer.override.compression.type: zstd
  producer.override.linger.ms: 20
  producer.override.batch.size: 200000
  max.queue.size: 30000
  max.batch.size: 8000
  log.mining.batch.size.min: 20000
  log.mining.batch.size.max: 200000
  log.mining.batch.size.default: 40000
  log.mining.query.filter.mode: in
  key.converter: org.apache.kafka.connect.json.JsonConverter
  key.converter.schemas.enable: false
  value.converter: org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable: false
  decimal.handling.mode: string
  topic.prefix: yms.data
  topic.creation.default.replication.factor: 3
  topic.creation.default.partitions: 3
  topic.creation.default.cleanup.policy: delete
  topic.creation.default.retention.ms: 21600000
  topic.creation.default.segment.ms: 7200000
  signal.data.collection: YMS.SRV_DEBEZIUM.DEBEZIUM_SIGNAL
  signal.enabled.channels: source
  signal.kafka.bootstrap.servers: "{{ .bootstrap_servers }}"
  log.mining.session.max.ms: 300000
  snapshot.max.threads: 4
  # message.key.columns: RPRT.BILING_OPER_SP:SERIALKEY, RPRT.BILING_OPER:SERIALKEY, OTD.THEOR_PICKDETAIL:ORDERKEY,UNITTYPEID, SDD.DELIVERY_HISTORY:ID_DEPARTMENT,ID_LINE,STATUS,ID_USER,ID_HEADER,ADDWHO,OLD_ID_CARRIER
  # transforms: RowIdToKey,WarehouseIdToKey
  transforms: WarehouseIdToKey,ExtractNewRecord, WarehouseIdToValue
  # transforms.RowIdToKey.type: org.apache.kafka.connect.transforms.ValueToKey 
  # transforms.RowIdToKey.fields: payload.after.row_id
  # transforms.RowIdToKey.predicate: TablesWithoutKey
  # predicates: TablesWithoutKey
  # predicates.TablesWithoutKey.type: org.apache.kafka.connect.transforms.predicates.TopicNameMatches
  # predicates.TablesWithoutKey.pattern: nexus-wms.data-test.SDD.DELIVERY_HISTORY
  transforms.WarehouseIdToKey.type: org.apache.kafka.connect.transforms.InsertField$Key
  transforms.WarehouseIdToKey.static.field: yms_id
  transforms.WarehouseIdToKey.static.value: "msk-dpro-ora581"
  transforms.WarehouseIdToValue.type: org.apache.kafka.connect.transforms.InsertField$Value
  transforms.WarehouseIdToValue.static.field: yms_id
  transforms.WarehouseIdToValue.static.value: "msk-dpro-ora581"
  transforms.ExtractNewRecord.type: io.debezium.transforms.ExtractNewRecordState
  transforms.ExtractNewRecord.add.fields: op,source.ts_ms,ts_ms,source.user_name,source.row_id,source.scn,source.commit_scn,source.txId,source.transaction
  transforms.ExtractNewRecord.delete.tombstone.handling.mode: rewrite
  custom.metric.tags: database=yms-msk-dpro-ora581
{{- end }}

[scholzj]

Normally, the connector just takes over the Kafka configuration from the Connect cluster. But Debezium has some specialties. I think you have various options - you can mount the JKS or PKCS12 stores, you can use the PEM files directly, you can use configuration providers. But I do no have any examples I'm afraid. Debezium project might have some examples, but not 100% sure.

[vsfomin]

This configuration works for me:
KafkaConnector:

schema.history.internal.producer.security.protocol: SSL
schema.history.internal.producer.ssl.keystore.type: JKS
schema.history.internal.producer.ssl.keystore.location: /mnt/kafka.keystore.jks
schema.history.internal.producer.ssl.keystore.password: "{{  .Values.keystore_password }}"
schema.history.internal.producer.ssl.truststore.location: /mnt/kafka.truststore.jks
schema.history.internal.producer.ssl.truststore.password: "{{  .Values.truststore_password }}"

schema.history.internal.consumer.security.protocol: SSL
schema.history.internal.consumer.ssl.keystore.type: JKS
schema.history.internal.consumer.ssl.keystore.location: /mnt/kafka.keystore.jks
schema.history.internal.consumer.ssl.keystore.password: "{{  .Values.keystore_password }}"
schema.history.internal.consumer.ssl.truststore.location: /mnt/kafka.truststore.jks
schema.history.internal.consumer.ssl.truststore.password: "{{  .Values.truststore_password }}"

KafkaConnect (partial)

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect-cluster-{{ .connectionName }}
  namespace: {{ $.Release.Namespace }}
  annotations:
    # use-connector-resources configures this KafkaConnect
    # to use KafkaConnector resources to avoid
    # needing to call the Connect REST API directly
    strimzi.io/use-connector-resources: "true"
spec:
  image: "{{ .debezium.image }}"
  serviceAccountName: {{ $.serviceAccountName }}
  replicas: 1
  bootstrapServers: "{{ .bootstrap_servers }}"
{{- if .KafkaConnect.tls }}
  tls:
{{ toYaml .KafkaConnect.tls | indent 4 }}
  {{- end }}
{{- if .KafkaConnect.authentication }}
  authentication:
{{ toYaml .KafkaConnect.authentication | indent 4 }}
 {{- end }}
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: connect-metrics
        key: metrics-config.yml
  jmxOptions: {}
  resources:
{{ toYaml .resources | indent 4 }}
  template:
    connectContainer:   #### Mount JKS Certs here! ###
      volumeMounts:
        - name: certs
          mountPath: /mnt
          readOnly: true
    pod:
      tmpDirSizeLimit: "5Gi"
      volumes:
      - name: connector-config
        secret:
          secretName: debezium-secret
      - name: certs
        secret:
          secretName: {{ $.Release.Namespace }}-jks