java - 来自 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对于该消费者组中的所有三个消费者具有相同的值(value)。

以下是我的设计细节:

我的 KafkaConsumerA将在一个消费者组中拥有三个消费者，每个消费者将在 topicA 上工作.

我的 KafkaConsumerB (类似于 KafkaConsumerA)将有两个消费者在不同的消费者组中，每个消费者都将在 topicB 上工作.

而这两个消费者 KafkaConsumerA和 KafkaConsumerB将在同一个盒子上运行，不同的消费者组相互独立。

Answer 1

我知道这是一个老问题，但考虑到这些天我们经常使用注释。所以添加了另一种风格的问题和答案。
我们面临同样的问题，但我们一直在同一个应用程序中的 2 个消费者中使用 @KafkaListener 注释，并且大多数属性直接注入(inject)

@KafkaListener(
topics = "${app.source}",
groupId = "${app.kafka.consumer.group-id}",
clientIdPrefix = "subscriber",
containerFactory = ListenerContainerFactory.AVRO_SPECIFIC,
errorHandler = "customConsumerAwareListenerErrorHandler"
)

我们的消费者有类似的实现，但连接到不同的主题，所以我们只是修改了“clientIdPrefix”，以便在实例化期间为他们提供唯一的值。所以最终的代码是
<-- 第一个消费者组件->

@KafkaListener(
topics = "${app.source}",
groupId = "${app.kafka.consumer.group-id}",
clientIdPrefix = "firstSubscriber",
containerFactory = ListenerContainerFactory.AVRO_SPECIFIC,
errorHandler = "customConsumerAwareListenerErrorHandler"
)

<-- 第二消费者组件->

@KafkaListener(
topics = "${app.source}",
groupId = "${app.kafka.consumer.group-id}",
clientIdPrefix = "secondSubscriber",
containerFactory = ListenerContainerFactory.AVRO_SPECIFIC,
errorHandler = "customConsumerAwareListenerErrorHandler"
)