?????? Kafka Consumer??????????????

[Even]

????????????????????????????????????




------------------ ???????? ------------------
??????:&nbsp;"zz zhang"<oliver7b...@gmail.com&gt;;
????????:&nbsp;2020??5??29??(??????) ????5:16
??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;"????"<jkill...@dingtalk.com&gt;;

????:&nbsp;Re: Kafka Consumer??????????????



??????maven-shade??????????????
????class????org.apache.kafka.common.serialization.ByteArrayDeserializer????????org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArrayDeserializer??????????????????Deserializer????shade????????????????org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.Deserializer??????????????

???? <jkill...@dingtalk.com.invalid&gt; ??2020??5??29?????? ????4:33??????
&gt;
&gt; ??????????????????jar??????
&gt;
&gt;
&gt; ------------------------------------------------------------------
&gt; ????????Even <452232...@qq.com&gt;
&gt; ??????????2020??5??29??(??????) 16:17
&gt; ????????user-zh <user-zh@flink.apache.org&gt;
&gt; ????????Kafka Consumer??????????????
&gt;
&gt; Hi??
&gt; ????????Kafka Consumer????????????
&gt; ????kafka&amp;nbsp;consumer&amp;nbsp;job ??????Flink session 
cluster????????????????????????????Flink per-job cluster 
????kafka??????????????????????????
&gt; ????flink??????1.10,kafka??????kafka_2.12-2.1.0????????consumer??????val 
data = env.addSource(new FlinkKafkaConsumer[String](topic, new 
SimpleStringSchema(), properties))
&gt; 2020-05-27&amp;nbsp;17:05:22
&gt; org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
&gt;&nbsp; at 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:811)
&gt;&nbsp; at 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:659)
&gt;&nbsp; at 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:639)
&gt;&nbsp; at 
org.apache.flink.streaming.connectors.kafka.internal.KafkaPartitionDiscoverer.initializeConnections(KafkaPartitionDiscoverer.java:58)
&gt;&nbsp; at 
org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.open(AbstractPartitionDiscoverer.java:94)
&gt;&nbsp; at 
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:505)
&gt;&nbsp; at 
org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
&gt;&nbsp; at 
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
&gt;&nbsp; at 
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:1007)
&gt;&nbsp; at 
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:454)
&gt;&nbsp; at 
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
&gt;&nbsp; at 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:449)
&gt;&nbsp; at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:461)
&gt;&nbsp; at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
&gt;&nbsp; at org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)
&gt;&nbsp; at java.lang.Thread.run(Thread.java:748)
&gt; Caused by: org.apache.kafka.common.KafkaException: 
org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArrayDeserializer
 is not an instance of org.apache.kafka.common.serialization.Deserializer
&gt;&nbsp; at 
org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:304)
&gt;&nbsp; at 
org.apache.kafka.clients.consumer.KafkaConsumer.<init&amp;gt;(KafkaConsumer.java:705)
&gt;&nbsp; ... 15 more



-- 
Best,
zz zhang