來自kafka消費者的InstanceAlreadyExistsException

Question

我正在與Kafka合作，並嘗試通過關注此article來設置消費者羣組。唯一的區別是我創建了自己的抽象類，處理器使設計更簡單。

下面是我的抽象類：

public abstract class Consumer implements Runnable { 
    private final Properties consumerProps; 
    private final String consumerName; 

    public Consumer(String consumerName, Properties consumerProps) { 
    this.consumerName = consumerName; 
    this.consumerProps = consumerProps; 
    } 

    protected abstract void shutdown(); 

    protected abstract void run(String consumerName, Properties consumerProps); 

    @Override 
    public final void run() { 
    run(consumerName, consumerProps); 
    } 
} 

下面是我KafkaConsumerA延伸上面抽象類：

public class KafkaConsumerA extends Consumer { 
    private KafkaConsumer<byte[], DataHolder> consumer; 

    public KafkaConsumerA(String consumerName, Properties consumerProps) { 
    super(consumerName, consumerProps); 
    } 

    @Override 
    public void shutdown() { 
    consumer.wakeup(); 
    } 

    @Override 
    protected void run(String consumerName, Properties consumerProps) { 
    // exception comes from below line from two of the threads and the remaining one thread works fine. 
    consumer = new KafkaConsumer<>(consumerProps); 
    List<String> topics = getTopicsBasisOnConsumerName(consumerName); 
    try { 
     consumer.subscribe(topics); 
     // Setup the schema config 
     Map<String, Object> config = new HashMap<>(); 
     config.put("urls", "https://abc.qa.host.com"); 

     GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config); 
     while (true) { 
     ConsumerRecords<byte[], DataHolder> records = consumer.poll(200); 
     for (ConsumerRecord<byte[], DataHolder> record : records) { 
      Map<String, Object> data = new HashMap<>(); 
      data.put("partition", record.partition()); 
      data.put("offset", record.offset()); 
      data.put("value", record.value()); 
      System.out 
       .println((Thread.currentThread().getId() % 3) + 1 + ": " + decoder.decode(record.value())); 
     } 
     } 
    } catch (WakeupException ex) { 
     ex.printStackTrace(); 
    } catch (Exception ex) { 
     ex.printStackTrace(); 
    } finally { 
     consumer.close(); 
    } 
    } 
} 

，下面將我的處理程序類：

// looks like something is wrong in this class 
public final class ConsumerHandler { 
    private final ExecutorService executorServiceProcess; 
    private final Consumer consumer; 
    private final List<Consumer> consumers = new ArrayList<>(); 

    public ConsumerHandler(Consumer consumer, int poolSize) { 
    this.executorServiceProcess = Executors.newFixedThreadPool(poolSize); 
    this.consumer = consumer; 
    for (int i = 0; i < poolSize; i++) { 
     consumers.add(consumer); 
     executorServiceProcess.submit(consumer); 
    } 
    } 

    public void shutdown() { 
    Runtime.getRuntime().addShutdownHook(new Thread() { 
     @Override 
     public void run() { 
     for (Consumer consumer : consumers) { 
      consumer.shutdown(); 
     } 
     executorServiceProcess.shutdown(); 
     try { 
      executorServiceProcess.awaitTermination(1000, TimeUnit.MILLISECONDS); 
     } catch (InterruptedException ex) { 
      Thread.currentThread().interrupt(); 
     } 
     } 
    }); 
    } 
} 

這裏我是從主類的消費者羣體中啓動我的所有消費者：

public static void main(String[] args) { 
    ConsumerHandler handlerA = 
     new ConsumerHandler(new KafkaConsumerA("KafkaConsumerA", getConsumerProps()), 3); 
    // run KafkaConsumerB here 

    handlerA.shutdown(); 
    // shutdown KafkaConsumerB here 
    } 

所以用這個 - 我的計劃是建立一個消費羣體有三個消費者KafkaConsumerA和所有三個訂閱相同的主題。

錯誤： -

每當我跑這個，貌似只有一個消費羣體在消費工作和其他兩名不起作用。我從控制檯上看到這個例外：

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=a97716e0-0e05-4938-8fa1-6b872cf24e34 
    at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900) ~[na:1.7.0_79] 
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324) ~[na:1.7.0_79] 
    at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[na:1.7.0_79] 
    at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58) ~[kafka-clients-0.10.0.0-SASL.jar:na] 
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:694) [kafka-clients-0.10.0.0-SASL.jar:na] 
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:587) [kafka-clients-0.10.0.0-SASL.jar:na] 
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:569) [kafka-clients-0.10.0.0-SASL.jar:na] 

我在這裏做什麼？ getConsumerProps()方法返回屬性對象，其中client.id和group.id與該消費者組中的所有三個消費者具有相同的值。

下面是我的設計細節：

• 我KafkaConsumerA將有3名消費者在消費羣體和每一個消費者將在topicA工作。
• 我的KafkaConsumerB（與KafkaConsumerA類似）將有兩個消費者在不同的消費羣體中，並且這些消費者中的每一個都將在topicB上工作。

而這兩個消費者KafkaConsumerA和KafkaConsumerB將運行在相互獨立的不同消費者組的相同盒子上。

Answer 1

Kafka正在嘗試註冊MBeans進行應用程序監視，並且正在使用client.id這樣做。正如您所說，您的抽象類中注入了屬性，併爲每個消費者注入相同的client.id和group.id組A。但是，你有不同的客戶，所以你應該給他們自己client.id，但保持相同group.id。這會將不同的客戶/消費者註冊到相同的消費者組中，並使它們一起工作，但不會與MBeans註冊發生衝突。