I’m doing a POC for Flink SQL. When reading from Kafka, I created the topic connection with the Flink DDL:

CREATE TABLE table_name ( ... ) WITH ( 'connector' = 'kafka' .... )

I’m reading from Confluent Kafka with security protocol:

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

The API Keys are in the DDL. I’d like to avoid that, How can I create this table with DDL without specifying these secret configs in it? Ideally I could inject these secret configs from in-code variables (fetched from secure place).

Hey, I think you can specify this using pyFlink TableAPI with table descriptor: https://nightlies.apache.org/flink/flink-docs-master/api/python/reference/pyflink.table/descriptors.html We do it like this:

TableDescriptor.for_connector('kafka')
            .schema(table_schema)
            .option('topic', kafka_topic_name)
            .option('properties.bootstrap.servers', bootstrap_server)
						#other options 
            .build()
        )

ah yeah, cool! Using Scala

TableDescriptor.forConnector("kafka")
      .schema(schema)
      .option('topic", ..)

Thanks!