wave I m still <https apache flink slack com archives C03G7 Apache Flink #troubleshooting

:wave: I'm still <trying to troubleshoot a connect...

Jon Slusher

12/10/2025, 7:15 PM

👋 I'm still trying to troubleshoot a connection from a Table API

kafka

source connector (using PyFlink) to our Kafka cluster using

SASL_SSL

and

SCRAM-SHA-512

. • Can anyone at least confirm that this should be possible? • I have some DEBUG logs to share that to me definitely point to the taskManager being the culprit. I see in the logs the jobManager is definitely connecting successfully to my Kafka brokers because it can see the topic I and partitions want to source. A taskManager is spawned shortly afterwards and that's what seems to file. • I'm using Flink 1.20.3, and the Flink operator in Kubernetes to manage the job • I'll post the configuration and a link to a condensed and redacted Gist of the DEBUG logs I captured for my connector in a thread

Jon Slusher

12/10/2025, 7:17 PM

Table API

kafka

source configuration:

Copy code

CREATE TABLE IF NOT EXISTS my_table (
  id BIGINT NOT NULL,
  release_id BIGINT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
    '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}\";',
    'properties.group.id' = 'flink.collection-aggregator',
    'format' = 'debezium-avro-confluent',
    'debezium-avro-confluent.url' = '{AVRO_REGISTRY}',
    'scan.startup.mode' = 'earliest-offset'
);
""")

Jon Slusher

12/10/2025, 7:18 PM

Condensed and sanitized DEBUG logs that show a successful connection from the jobManager and failure from the taskManager

H.P. Grahsl

12/10/2025, 8:00 PM

A bit hard to say based on the logs you attached. But I think the exception in that file you shared might be about something else not this table definition above that uses kafka as a source.

H.P. Grahsl

12/10/2025, 8:01 PM

The way I read it is that it should be about some aggregate topic that your pyflink job tries to write so related to sink?

Jaehyeon Kim

12/10/2025, 8:05 PM

Can you check the connector doc? https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/table/kafka/#security 1. It doesn't have SSL config 2. The class path of the login module is for the SQL client. Are you using that?

Jon Slusher

12/10/2025, 8:10 PM

I think I see what I did now. The source connector is working fine. It's my second

upsert-kafka

connector that wasn't working, and it's because I never set authentication for it 🤦‍♂️ :

Copy code

CREATE TABLE IF NOT EXISTS my_count (
    example_id BIGINT,
    example_count BIGINT,
    PRIMARY KEY (release_id) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'flink.my_count_topic',
    '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}\";',
    'key.format' = 'json',
    'value.format' = 'json'
);

H.P. Grahsl

12/10/2025, 8:10 PM

👍 like I assumed. something about the sink maybe 🙂 good luck.

Jon Slusher

12/10/2025, 8:10 PM

Setting that I'm getting a different error about a topic not found, but I can work through that one I think. Thanks for the reply!

H.P. Grahsl

12/10/2025, 8:11 PM

yeah this could be because e.g. you have no auto creation configured or allowed for topics on the broker.

👍 1

Open in Slack

Previous Next