Flink kafka connector with JAAS configurations crashed

[sundy]

Hi ,all 

I use the code below to set kafka JASS config,   the serverConfig.jasspath is  /data/apps/spark/kafka_client_jaas.conf,   but on flink standalone deployment, it crashs. I am sure the kafka_client_jass.conf is valid, cause other applications(Spark streaming) are still working fine with it. So I think it may be not the problem caused by kafka 0.10 client.

System.setProperty("java.security.auth.login.config", serverConfig.jasspath);
properties.setProperty("security.protocol", "SASL_PLAINTEXT");
properties.setProperty("sasl.mechanism", "PLAIN");

Exceptions msgs are:

org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:717)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:597)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:579)
	at org.apache.flink.streaming.connectors.kafka.internal.Kafka09PartitionDiscoverer.initializeConnections(Kafka09PartitionDiscoverer.java:56)
	at org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.open(AbstractPartitionDiscoverer.java:91)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:422)
	at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
	at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:393)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:254)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.KafkaException: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is /data/apps/spark/kafka_client_jaas.conf
	at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:94)
	at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:93)
	at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:51)
	at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:84)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:657)
	... 11 more
Caused by: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is /data/apps/spark/kafka_client_jaas.conf
	at org.apache.kafka.common.security.JaasUtils.defaultJaasConfig(JaasUtils.java:85)
	at org.apache.kafka.common.security.JaasUtils.jaasConfig(JaasUtils.java:67)
	at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:85)
	... 15 more

File content looks like below:

KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password=“xxxxxxxxxxx"; };

It seems like the kafka_client_jaas.conf file has been read, but the KafkaClient entry could not be resolved. That’s very strange, other applications with the same config are working fine. And I wrote a simple Java code to test the the file, it works fine too.

public static void main(String[] args) {

  Map<String, String> maps = new HashMap<>();

  System.setProperty("java.security.auth.login.config", "/data/apps/spark/kafka_client_jaas.conf");

  Configuration jassConfig = JaasUtils.jaasConfig(LoginType.CLIENT, maps);

  AppConfigurationEntry object[] = jassConfig.getAppConfigurationEntry("KafkaClient");

  for(AppConfigurationEntry entry : object){

    System.out.println(entry.getOptions());

  }

}

 

[Nico Kruber]

Hi,
I'm no expert on Kafka here, but as the tasks are run on the worker
nodes (where the TaskManagers are run), please double-check whether the
file under /data/apps/spark/kafka_client_jaas.conf on these nodes also
contains the same configuration as on the node running the JobManager,
i.e. an appropriate entry for 'KafkaClient'.


Regards
Nico

On 13/03/18 08:42, sundy wrote:

>
> Hi ,all 
>
> I use the code below to set kafka JASS config,   the
> serverConfig.jasspath is  /data/apps/spark/kafka_client_jaas.conf,   but
> on flink standalone deployment, it crashs. I am sure the
> kafka_client_jass.conf is valid, cause other applications(Spark
> streaming) are still working fine with it. So I think it may be not the
> problem caused by kafka 0.10 client.
>
> System.setProperty("java.security.auth.login.config", serverConfig.jasspath);
> properties.setProperty("security.protocol", "SASL_PLAINTEXT");
> properties.setProperty("sasl.mechanism", "PLAIN");
>
>
> Exceptions msgs are:
>
> org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
> at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:717)
> at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:597)
> at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:579)
> at org.apache.flink.streaming.connectors.kafka.internal.Kafka09PartitionDiscoverer.initializeConnections(Kafka09PartitionDiscoverer.java:56)
> at org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.open(AbstractPartitionDiscoverer.java:91)
> at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:422)
> at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
> at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:393)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:254)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: org.apache.kafka.common.KafkaException: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is /data/apps/spark/kafka_client_jaas.conf
> at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:94)
> at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:93)
> at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:51)
> at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:84)
> at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:657)
> ... 11 more
> Caused by: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is /data/apps/spark/kafka_client_jaas.conf
> at org.apache.kafka.common.security.JaasUtils.defaultJaasConfig(JaasUtils.java:85)
> at org.apache.kafka.common.security.JaasUtils.jaasConfig(JaasUtils.java:67)
> at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:85)
> ... 15 more
>
>
>
> File content looks like below:
>
> KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule
> required username="admin" password=“xxxxxxxxxxx"; };
>
> It seems like the kafka_client_jaas.conf file has been read, but the
> KafkaClient entry could not be resolved. That’s very strange, other
> applications with the same config are working fine. And I wrote a simple
> Java code to test the the file, it works fine too.
>
>
> public static void main(String[] args) {
>   Map<String, String> maps = new HashMap<>();
>   System.setProperty("java.security.auth.login.config",
> "/data/apps/spark/kafka_client_jaas.conf");
>   Configuration jassConfig = JaasUtils.jaasConfig(LoginType.CLIENT, maps);
>   AppConfigurationEntry object[] =
> jassConfig.getAppConfigurationEntry("KafkaClient");
>   for(AppConfigurationEntry entry : object){
>     System.out.println(entry.getOptions());
>   }
> }
>
>
>
>
>  
>
>
>