flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ForwardXu" <x1q...@qq.com>
Subject 回复:kafka consumer exception
Date Wed, 20 Feb 2019 07:23:37 GMT
董鹏,你好:


   你这个问题可能多半是你在kafka consumer配置中配置了client-id,然后flink多线程执行的时候用的是一样client-id向kafka请求消费数据导致。具体问题你可参看以下jira链接:
https://issues.apache.org/jira/browse/KAFKA-3992。如果是配置了client-id可以去掉留空。这样kafka会为每一个线程从新生成一个clientid,"consumer"
+  自增id。




前进


------------------ 原始邮件 ------------------
发件人: "董鹏"<dongpeng@imdada.cn>;
发送时间: 2019年2月20日(星期三) 下午3:02
收件人: "user-zh"<user-zh@flink.apache.org>;

主题: kafka consumer exception



flink大神你们好,在使用flink on kafka(1.0版本) 遇到如下异常:
不影响job,不影响结果,对于这个异常偶尔打出,你们是否有遇到这个问题呢?


[org.apache.kafka.common.utils.AppInfoParser] [AppInfoParser.java:60] - Error registering
AppInfo mbean
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=consumer-31
  at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
  at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)
  at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966)
  at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900)
  at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324)
  at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)
  at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58)
  at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:709)
  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:58)
  at org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.open(AbstractPartitionDiscoverer.java:94)
  at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:469)
  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:424)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:290)
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
  at java.lang.Thread.run(Thread.java:748)
Mime
  • Unnamed multipart/alternative (inline, 8-Bit, 0 bytes)
View raw message