Kafka CommitFailedException wyjątek konsument

Question

Po tworzenie wielu konsumentów (używając Kafka 0.9 Java API), a każdy wątek rozpoczęty, ja otrzymuję następujący wyjątek

Consumer has failed with exception: org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance 
class com.messagehub.consumer.Consumer is shutting down. 
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance 
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:546) 
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:487) 
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:681) 
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:654) 
at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:167) 
at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133) 
at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107) 
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.onComplete(ConsumerNetworkClient.java:350) 
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:288) 
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:303) 
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:197) 
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:187) 
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:157) 
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:352) 
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:936) 
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:905) 

a następnie rozpocząć spożywanie wiadomość normalnie, chciałbym wiedzieć, co powoduje ten wyjątek w celu jego usunięcia.

Answer 1

Jej problem z równoważeniem grupy konsumentów wynika z błędów. Czy możesz nam powiedzieć, z ilu utworzono temat partycji? i ilu konsumentów pracuje? Czy należą do tej samej grupy?

Answer 2

Spróbuj również dostosować następujące parametry:

• heartbeat.interval.ms - To mówi Kafka czekać określoną ilość milisekund przed rozważyć konsument będzie uważany za „martwy”
• max.partition.fetch.bytes - Ograniczy to ilość wiadomości (do), które konsument otrzyma podczas odpytywania.

Zauważyłem, że ponowne zrównoważenie występuje, jeśli konsument nie zobowiązuje się do Kafki przed upływem czasu bicia serca. Jeśli zatwierdzenie nastąpi po przetworzeniu komunikatów, czas ich przetworzenia określi te parametry. Zmniejszenie liczby wiadomości i zwiększenie czasu bicia serca pomoże uniknąć ponownego zrównoważenia.

Należy również rozważyć użycie większej liczby partycji, więc będzie więcej wątków przetwarzających dane, nawet z mniejszą liczbą wiadomości na sondę.

Napisałem tę małą aplikację do testów. Mam nadzieję, że to pomoże.

https://github.com/ajkret/kafka-sample

UPDATE

Kafka 0.10.x oferuje teraz nowy parametr kontrolować ilość otrzymanych wiadomości: - max.poll.records - Maksymalna liczba rekordów zwracana w pojedyncze wywołanie poll().

UPDATE

Kafka oferuje możliwość pauzie kolejce. Podczas gdy kolejka jest wstrzymana, możesz przetwarzać wiadomości w oddzielnym wątku, co pozwala zadzwonić pod numer KafkaConsumer.poll(), aby wysłać bicie serca. Następnie wywołaj KafkaConsumer.resume() po zakończeniu przetwarzania. W ten sposób można złagodzić problemy związane z przywracaniem równowagi z powodu braku wysyłania pulsu. Oto zarys tego, co możesz zrobić:

while(true) { 
    ConsumerRecords records = consumer.poll(Integer.MAX_VALUE); 
    consumer.commitSync(); 

    consumer.pause(); 
    for(ConsumerRecord record: records) { 

     Future<Boolean> future = workers.submit(() -> { 
      // Process 
      return true; 
     }); 


     while (true) { 
      try { 
       if (future.get(1, TimeUnit.SECONDS) != null) { 
        break; 
       } 
      } catch (java.util.concurrent.TimeoutException e) { 
       getConsumer().poll(0); 
      } 
     } 
    } 

    consumer.resume(); 
}