java 服务器

kafka 并发数配置过程中踩到的坑 InstanceAlreadyExistsException

2017-07-05 13:09:15.460 [kafka_spout:7-MultipleThreadSpoutExecutors] WARN o.a.kafka.common.utils.AppInfoParser - Error registering AppInfo mbean

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=gx-test-20170629
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:57)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:640)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:512)
at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:73)
at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumer(DefaultKafkaConsumerFactory.java:69)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:284)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:222)
at org.springframework.kafka.listener.KafkaMessageListenerContainer.doStart(KafkaMessageListenerContainer.java:179)
at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:204)
at org.springframework.kafka.listener.ConcurrentMessageListenerContainer.doStart(ConcurrentMessageListenerContainer.java:126)
at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:204)
at com.mapbar.stream.qingqi.core.spout.KafkaSpout.open(KafkaSpout.java:74)
at com.alibaba.jstorm.task.execute.spout.SpoutExecutors.init(SpoutExecutors.java:142)
at com.alibaba.jstorm.task.execute.spout.MultipleThreadSpoutExecutors.init(MultipleThreadSpoutExecutors.java:64)
at com.alibaba.jstorm.task.execute.BaseExecutors.initWrapper(BaseExecutors.java:154)
at com.alibaba.jstorm.task.execute.spout.MultipleThreadSpoutExecutors.run(MultipleThreadSpoutExecutors.java:76)
at com.alibaba.jstorm.callback.AsyncLoopRunnable.run(AsyncLoopRunnable.java:95)
at java.lang.Thread.run(Thread.java:745)

上面是本人在使用spring kafka中所遇到的问题，针对此问题做一个记录，整理到此处

出现上述问题的原因：

ConcurrentMessageListenerContainer factory = new ConcurrentMessageListenerContainer(cf, containerProps);
factory.setConcurrency(kafkaConfig.getConcurrencySize());
如果使用了ConcurrentMessageListenerContainer 的实现，并且配置了并发度大于1，同时配置了kafka的 client.id属性则会出现上述问题，而当你配置为1的时候不会出现上述log
解决方式：不配置client.id这一项，kakfa中会默认为多个线程生成id
详细解析：
  org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:640)
  从error log中可以看出

       在调用AppInfoParser.registerAppInfo方法时出现的异常

　　

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)

　　从堆栈信息中定位到 Repository.addMBean()

截取其中一部分代码

上图中可以得知，一个clientId对应一个，不可重复

最后找到KafkaConsumer，发现是因为自己配置了client.id导致的，如果不配置的话

会为每一个线程生成一个clientid，"consumer" +  自增id，原子性递增

看到这里，发现了一个问题，如果自己不配置client.id的话，那从config里取出来的数据直接判断的length，猜测是有默认配置，当自己不配置的时候给赋值为"" 空串，于是又查了一下代码来验证

创建consumer对象的时候，会创建ConsumerConfig 配置，new ConsumerConfig()会调用父类构造方法

ConsumerConfig(Map<?, ?> props) {
    super(CONFIG, props);
}

AbstractConfig

ConsumerConfig中有static静态初始化块，来初始化 ConfigDef

把client.id赋值为""