Kafka消费者停止消费，如果其中一个经纪人变得不可用

Question

我在一台Windows主机上安装了两个Kafka 2.1.0代理。默认复制因子设置为2。所有其他设置均为默认设置。

networkClient:[Consumer ClientID=Consumer-1,GroupID=SOUT]无法建立到节点-2(/192.168.0.1:19092)的连接。代理可能不可用。

消费者：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Properties;

public final class PumpToConsoleSimple {

  private static final Duration pollTimeout = Duration.ofSeconds(10);

  public static void main(String[] args) {
    final Properties consumerProperties = new Properties();

    consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "sout");
    consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092,192.168.0.1:19092");
    consumerProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);

    try (final KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(consumerProperties)) {
      kafkaConsumer.subscribe(Collections.singleton("test"), new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
          //do nothing
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
          System.out.println("Partitions were assigned");
          kafkaConsumer.seekToBeginning(partitions);
        }
      });
      while (true) {
        final ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(pollTimeout);
        consumerRecords.forEach(r -> System.out.println(r.value()));
        kafkaConsumer.commitSync();
      }
    }
  }
}

一个制作人：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.locks.LockSupport;

public final class OnceInASecondProducerSimple {
  public static void main(String[] args) {
    final Properties producerProperties = new Properties();
    producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092,192.168.0.1:19092");

    long counter = 0;
    while (true) {
      try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProperties)) {
        producer.send(new ProducerRecord<>("test", "msg" + counter++));
      }
      LockSupport.parkNanos(Duration.ofSeconds(1).getNano());
    }
  }
}

Answer 1

通过kafka-topics脚本检查偏移量主题__consumer_offsets状态。失败的代理必须是组的协调器，并且__consumer_offsets的复制因子可能为1，因此使用者找不到协调器。即使您重新启动消费者，它仍然无法查找协调器。

在您的情况下，可以增加__consumer_offsets的复制因子，然后重试看看它是否按预期工作。