Потребитель Кафки прекращает потребление, если один из брокеров становится недоступным - PullRequest
Новая
       10

Потребитель Кафки прекращает потребление, если один из брокеров становится недоступным

0 голосов
/ 30 января 2019

У меня установлено два брокера Kafka 2.1.0 на одном хосте Windows.Коэффициент репликации по умолчанию установлен равным 2. Все остальные настройки являются настройками по умолчанию.

Производитель может записывать сообщения в журнал, даже если я отключаю одного из посредников.Но потребитель перестает потреблять сообщения в этом случае.Даже если я перезапущу его, ему не будут назначены разделы.Он просто записывает в журнал это предупреждение:

main - org.apache.kafka.clients.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());
    }
  }
}

Потребитель продолжает работу только после повторного запуска брокера.

Что я пропустил?Как получить высокую доступность для потребителя Kafka?

1 Ответ

0 голосов
/ 30 января 2019

Проверить смещение темы __consumer_offsets состояние с помощью kafka-topics сценария.Отказавший брокер должен быть координатором для вашей группы, а коэффициент репликации __consumer_offsets может быть равен 1, поэтому потребитель не может найти координатора.Даже если вы перезапустите потребителя, он все равно не сможет найти координатор.

В вашем случае вы можете увеличить коэффициент репликации __consumer_offsets и повторить попытку, чтобы проверить, работает ли он должным образом.

...