My Debezium connector, which is working:

authentication:
    type: scram-sha-512
    username: "kc-test"
    passwordSecret:
      secretName: strimzi-connect-secrets
      password: kc-test-password
  bootstrapServers: <http://strimzi-kafka-bootstrap.staging.example.com:9096|strimzi-kafka-bootstrap.staging.example.com:9096>

I have this set in my Job parameters:

'connector' = 'kafka',
    'topic' = 'debezium.sdlc-example.example.my_table',
    'properties.bootstrap.servers' = '{BOOTSTRAP}',
    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.mechanism' = 'SCRAM-SHA-512',
    'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username=\"{KAFKA_SASL_USERNAME}\" password=\"{KAFKA_SASL_PASSWORD}\";',

With this configured as the bootstrap server:

BOOTSTRAP = "<http://strimzi-kafka-bootstrap.staging.example.com:9096|strimzi-kafka-bootstrap.staging.example.com:9096>"

I see several of these in the TaskManager logs:

2025-12-05 21:05:24,581 WARN  org.apache.flink.kafka.shaded.org.apache.kafka.clients.NetworkClient [] - [Producer clientId=producer-1] Bootstrap broker <http://strimzi-kafka-bootstrap.staging.example.com:9096|strimzi-kafka-bootstrap.staging.example.com:9096> (id: -1 rack: null) disconnected
2025-12-05 21:05:25,643 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.NetworkClient [] - [Producer clientId=producer-1] Node -1 disconnected.
2025-12-05 21:05:25,644 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.NetworkClient [] - [Producer clientId=producer-1] Cancelled in-flight API_VERSIONS request with correlation id 39 due to node -1 being disconnected (elapsed time since creation: 8ms, elapsed time since send: 8ms, request timeout: 30000ms)

Looking up this error suggests there is no connection being made at all, as if there was a network partition, but I see earlier in the logs before the TaskManager pod starts that the JobManager pod seems to at least claim that it's seeing partitions of the topic I want to source:

org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator [] - Discovered new partitions: [debezium.sdlc-example.example.my_table-61, debezium.sdlc-example.example.my_table-32, debezium.sdlc-example.example.my_table-65, debezium.sdlc-example.example.my_table-36, debezium.sdlc-example.example.my_table-3,

I think these logs in the brokers are related, but again I don't know why the jobManager would connect and authenticate but the taskManager isn't:

23:06:29,132 INFO [SocketServer listenerType=ZK_BROKER, nodeId=2] Failed authentication with /172.26.15.0 (channelId=172.24.152.2:9096-172.26.15.0:8774-864) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-2-ListenerName(SCRAM-9096)-SASL_SSL-8]

Also I'm confused by what looks like a successful connection on the jobManager. It's able to at least find the topic and its partitions. Doesn't that mean that authentication config is working?

It is not likely that authentication works on the JobManager but not TaskManagers.

I wouldn't think so either, but how is the jobManager seeing the topic and its partitions before spawning the taskManager?

I added the following after retrieving the truststore certificates from my Kafka cluster and adding it to my container:

'properties.ssl.truststore.location' = '/opt/flink/certs/ca.p12',
    'properties.ssl.truststore.password' = '{CA_TRUSTSTORE_SECRET}',
    'properties.ssl.truststore.type' = 'PKCS12',

I'm getting a different error now, but I verified that the certificate exists in the location I specified:

org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target

But for whatever reason, I'm supplying what I assume are the proper credentials and I'm still getting an SSL error. I tried with the cluster's PEM cert:

'properties.ssl.truststore.location' = '/etc/ssl/certs/ca.crt',
    'properties.ssl.truststore.type' = 'PEM',

but I'm getting the same error:

Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target

The

ca.crt

file was exported from the cluster and I'm wondering if it needs to have intermediary certificates added or something :\

Or does it need to be added to

/etc/ssl/certs/java/cacerts

maybe?