Hunter Kempf created FLINK-15111:
------------------------------------
Summary: java.lang.RuntimeException for
KafkaPartitionDiscoverer.getAllPartitionsForTopics(KafkaPartitionDiscoverer.java:80)
when using SASL_SSL
Key: FLINK-15111
URL: https://issues.apache.org/jira/browse/FLINK-15111
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka
Affects Versions: 1.9.0
Environment: Flink Version 1.9.0
Scala Version 2.11.12
Kafka Cluster Version 2.3.0
Reporter: Hunter Kempf
Goal: I am trying to connect a flink job I made to a remote kafka cluster that
has 3 partitions and requires authentication with SASL_SSL.
Background: I have tested my job against a kafka cluster topic running on my
localhost that has one partition as well as one that had three partitions and
it works to read and write to the local kafka.
Problem: When I attempt to connect to a topic that has multiple partitions and
SASL_SSL I get the following error (for reference topicName is the name of the
topic I am trying to consume). Weirdly I dont have any issues when I am trying
to produce to a remote multi-partition topic.
```
{{java.lang.RuntimeException: topicName
at
org.apache.flink.streaming.connectors.kafka.internal.KafkaPartitionDiscoverer.getAllPartitionsForTopics(KafkaPartitionDiscoverer.java:80)
at
org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.discoverPartitions(AbstractPartitionDiscoverer.java:131)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:508)
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:529)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:393)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
at java.lang.Thread.run(Thread.java:748)}}
```
My Consumer Code looks like this :
```
{{ def defineKafkaDataStream[A: TypeInformation](topic: String,
env: StreamExecutionEnvironment,
SASL_username:String,
SASL_password:String,
kafkaBootstrapServer: String =
"localhost:9092",
zookeeperHost: String =
"localhost:2181",
groupId: String = "test"
)(implicit c: JsonConverter[A]):
DataStream[A] = \{
val properties = new Properties()
properties.setProperty("bootstrap.servers", kafkaBootstrapServer)
properties.setProperty("security.protocol" , "SASL_SSL")
properties.setProperty("sasl.mechanism" , "PLAIN")
val jaasTemplate = "org.apache.kafka.common.security.plain.PlainLoginModule
required username=\"%s\" password=\"%s\";"
val jaasConfig = String.format(jaasTemplate, SASL_username, SASL_password)
properties.setProperty("sasl.jaas.config", jaasConfig)
properties.setProperty("group.id", "MyConsumerGroup")
env
.addSource(new FlinkKafkaConsumer(topic, new
JSONKeyValueDeserializationSchema(true), properties))
.map(x => x.convertTo[A](c))
}}}
```
Is there some property I should be setting that I am not? Since it worked fine
in Local I assume this is a bug with the interaction of the partition
discoverer when authenticating with SASL_SSL but if it is something I can fix
with a setting please let me know what to change
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
