Flink Kafka Table API for python with JAAS

[Sweta Kalakuntla]

Hi,

I am using Flink 1.11.2 version Python Table API to connect to Kafka Topic using SASL protocol but it fails with the following error. I tried the same properties in Flink java version, and I am able to connect. Has anyone faced this issue and how did you resolve it?

Error:

Caused by: javax.security.auth.login.LoginException: unable to find LoginModule class: org.apache.kafka.common.security.plain.PlainLoginModule

Kafka connection:

kafka = Kafka()\
        .version("universal") \
        .topic("test_topic")\
        .property("group.id", "consumer_group")\
        .property("security.protocol", "SASL_PLAINTEXT")\
        .property("sasl.mechanism", "PLAIN")\
        .property("bootstrap.servers",
                  "<remoteIP>:9093")\
        .property("sasl.jaas.config",
                  "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" "
                  "password=\"abc\";")\
        .start_from_latest()

Thank you,

SKala

--

[Timo Walther]

Hi,

are you using the SQL jars or do you build the dependency jar file
yourself? It might be the case that the SQL jar for Kafka does not
include this module as the exception indicates. You might need to build
a custom Kafka jar with Maven and all dependencies you need. (including
correct META-INF/services entries).

I hope this helps.

Regards,
Timo

On 10.11.20 05:11, Sweta Kalakuntla wrote:

> Hi,
>
> I am using Flink 1.11.2 version Python Table API to connect to Kafka
> Topic using SASL protocol but it fails with the following error. I tried
> the same properties in Flink java version, and I am able to connect. Has
> anyone faced this issue and how did you resolve it?
>
> Error:
>
> |Caused by: javax.security.auth.login.LoginException: unable to find
> LoginModule class: org.apache.kafka.common.security.plain.PlainLoginModule|
>
> Kafka connection:
>
> |kafka = Kafka()\ .version("universal") \ .topic("test_topic")\
> .property("group.id <http://group.id/>", "consumer_group")\
> .property("security.protocol", "SASL_PLAINTEXT")\
> .property("sasl.mechanism", "PLAIN")\ .property("bootstrap.servers",
> "<remoteIP>:9093")\ .property("sasl.jaas.config",
> "org.apache.kafka.common.security.plain.PlainLoginModule required
> username=\"user\" " "password=\"abc\";")\ .start_from_latest()|
>
>
> Thank you,
> SKala
>
> --
>
> BandwidthBlue.png
>
>
>
> Sweta Kalakuntla•Software Engineer
>
> 900 Main Campus Drive, Raleigh, NC 27606
>
>
> m:216-702-1653
>
> e: [hidden email] <http://bandwidth.com>
>

[Xingbo Huang]

Hi,

You can use the following API to add all the dependent jar packages you need:

table_env.get_config().get_configuration().set_string("pipeline.jars", "file:///my/jar/path/connector.jar;file:///my/jar/path/json.jar")

For more related content, you can refer to the pyflink doc[1]

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/python/table-api-users-guide/python_table_api_connectors.html#download-connector-and-format-jars


Best,
Xingbo

Timo Walther <[hidden email]> 于2020年11月10日周二 下午10:46写道：

Hi,

are you using the SQL jars or do you build the dependency jar file
yourself? It might be the case that the SQL jar for Kafka does not
include this module as the exception indicates. You might need to build
a custom Kafka jar with Maven and all dependencies you need. (including
correct META-INF/services entries).

I hope this helps.

Regards,
Timo

On 10.11.20 05:11, Sweta Kalakuntla wrote:
> Hi,
>
> I am using Flink 1.11.2 version Python Table API to connect to Kafka
> Topic using SASL protocol but it fails with the following error. I tried
> the same properties in Flink java version, and I am able to connect. Has
> anyone faced this issue and how did you resolve it?
>
> Error:
>
> |Caused by: javax.security.auth.login.LoginException: unable to find
> LoginModule class: org.apache.kafka.common.security.plain.PlainLoginModule|
>
> Kafka connection:
>
> |kafka = Kafka()\ .version("universal") \ .topic("test_topic")\
> .property("group.id <http://group.id/>", "consumer_group")\
> .property("security.protocol", "SASL_PLAINTEXT")\
> .property("sasl.mechanism", "PLAIN")\ .property("bootstrap.servers",
> "<remoteIP>:9093")\ .property("sasl.jaas.config",
> "org.apache.kafka.common.security.plain.PlainLoginModule required
> username=\"user\" " "password=\"abc\";")\ .start_from_latest()|
>
>
> Thank you,
> SKala
>
> --
>
> BandwidthBlue.png
>
>       
>
> Sweta Kalakuntla•Software Engineer
>
> 900 Main Campus Drive, Raleigh, NC 27606
>
>
> m:216-702-1653
>
> e: [hidden email] <http://bandwidth.com>
>