Kafka - Сбой фиксации на разделе MYTOPIC-X со смещением 0 - PullRequest
Новая
       7

Kafka - Сбой фиксации на разделе MYTOPIC-X со смещением 0

0 голосов
/ 17 февраля 2019

когда система находится в режиме ожидания и включено автоматическое смещение фиксации, время от времени я получаю очень странное сообщение об ошибке: 2019-02-15 16:31:42,742 ERROR[pool-9-thread-1] org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=ClientVerticle_Thread[pool-9-thread-1,5,main], groupId=mygroupid] Offset commit failed on partition MYYOPIC-5 at offset 0: The request timed out.', u'[Consumer clientId=ClinetVerticle_Thread[pool-9-thread-1,5,main], groupId=mygroupid] Offset commit failed on partition MYTOPIC-5 at offset 0: The request timed out Некоторые данные о настройке, которые воспроизводят эту ошибку:

  • автокоммит включен
  • Эта ошибка возникает, только если смещение равно 0 (сообщения не посещали раздел)
  • Эта ошибка возникает, только если к теме подключено более 1 группы потребителей (в данном случае 2 разных группы потребителей)
  • Количество разделов тем: 6
  • Количество потоков потребителей (для каждой группы потребителей): 3
  • Настройка потребителя: 

    config.put("client.id", InetAddress.getLocalHost().getHostAddress() + "_" + clientId + "_" + Thread.currentThread());
                    config.put("group.id", consumerGroupName);
                    config.put("bootstrap.servers", clusterUrl);
                    config.put("auto.offset.reset", "latest");
                    config.put("heartbeat.interval.ms", 3000);
                    config.put("session.timeout.ms", 10000);
                    config.put("request.timeout.ms", 10000 + 3000);
                    config.put("enable.auto.commit", true);
                    config.put("auto.commit.interval.ms", 2000);
                    config.put("key.deserializer", StringDeserializer.class.getName());
                    config.put("value.deserializer", StringDeserializer.class.getName());
                    config.put("fetch.min.bytes", 1);
                    config.put("max.partition.fetch.bytes", 1048576);
                    config.put("fetch.max.wait.ms", 500);

СпасибоВы

...