我正面临一个与Kafka有关的问题。
我现在的服务( Producer
)将信息发送到Kafka主题( events
). 服务正在使用 kafka_2.12 v1.0.0
,用java编写。
我正在尝试将它与 spark-streaming
作为一个 Consumer
服务(这里使用kafka_2.11 v0.10.0,用scala编写)
消息已从成功发送 Producer
Kafka主题。但是,我总是收到下面的错误堆栈:
Exception in thread "main" org.apache.kafka.common.errors.InconsistentGroupProtocolException: The group member's supported protocols are incompatible with those of existing members.
at ... run in separate thread using org.apache.spark.util.ThreadUtils ... ()
at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:577)
at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:571) at com.jj.streaming.ItemApp$.delayedEndpoint$com$jj$streaming$ItemApp$1(ItemApp.scala:72)
at com.jj.streaming.ItemApp$delayedInit$body.apply(ItemApp.scala:12)
at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12) at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.collection.immutable.List.foreach(List.scala:381)
at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35) at scala.App$class.main(App.scala:76)
at com.jj.streaming.ItemApp$.main(ItemApp.scala:12)
at com.jj.streaming.ItemApp.main(ItemApp.scala)
我不知道根本原因。我怎样才能解决这个问题?
1条答案
按热度按时间iqjalb3h1#
在我的配置中,当我尝试向集群添加一个使用不同于以前分区分配策略的使用者时,就会发生这种情况。
例如:
partition.assignment.strategy=org.apache.kafka.clients.consumer.randomaccessassignor
混合或默认为:
partition.assignment.strategy=org.apache.kafka.clients.consumer.rangeassignor