Hi, I am using FlinkSQL to submit a job that uses upsert-kafka connector as sink to publish data to Confluent Topic but getting error that topic is not found. Any advice to resolve this

Caused by: <http://org.apache.flink.kafka.shaded.org|org.apache.flink.kafka.shaded.org>.apache.kafka.common.errors.TimeoutException: Topic flink not present in metadata after 60000 ms.

This usually happens when there's a connectivity issue

anything I am missing in below ?

Flink SQL> CREATE TABLE KafkaSinkTable (
>    id INT primary key not enforced,
>    net_score INT
>  ) WITH (
>    'connector' = 'upsert-kafka',
>    'topic' = 'flink',
>    'properties.bootstrap.servers' = 'xxxx-xxxxxx.ap-south-1.aws.confluent.cloud:9092',
>    'key.format' = 'json' , 'value.format' = 'json' );

You need this;

'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="API_KEY" password="API_SECRET";',

See https://developer.confluent.io/courses/apache-flink/deploying-an-etl-pipeline-using-flink-sql-exercise/#create-a-table-backed-by-kafka for more details.

Thanks @David Anderson Sir, after making the changes as suggested, now getting below error

javax.security.auth.login.LoginException: No LoginModule found for org.apache.kafka.common.security.plain.PlainLoginModule

Make sure that you’re using the right classpath, since Flink shades it by default. See https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/kafka/#security at the bottom of that section

Thanks @Martijn Visser Sir, it worked