Předpokládejme, že Kafka, 1 partition, 2 consumers.(2. spotřebitel je nečinný)
Předpokládejme, že 1. jedna spotřebované zprávu, jde to zvládnout s 3 ostatní služby a náhle hole na jeden z nich, a slečna Kafka je časový limit.
Bude Kafka potvrdí oddíl 2. spotřebitele a zpráva bude dvojnásob zpracovány (předpokládejme, že 1. jeden nakonec uspět)?
Co když Kafka je spotřebitel zpracovává zprávu příliš dlouho? Bude Kafka potvrdí tento oddíl do jiného spotřebitele a zpráva bude dvojnásob zacházet?
Ano, to je správné. Pokud Kafka spotřebitele trvá příliš dlouho, aby zvládnout zprávu a následné poll() je zpožděn, Kafka bude znovu jmenovat tento oddíl do jiného spotřebitele a zpráva bude zpracována znovu (a znovu).
Pro větší přehlednost, nejprve se musíme rozhodnout, a definovat, Jak dlouho je příliš dlouhý?'.
Tohle je definována vlastnost max.poll.interval.ms. Od docs,
Maximální zpoždění mezi vyvolání poll() při použití spotřebitele vedení skupiny. Tato místa horní mez na množství času, že spotřebitel může být v nečinnosti, než načítání více záznamů. Pokud poll() není volána před uplynutím tohoto limitu, pak spotřebitel je považován za neúspěšný, a skupina bude přehodnotit s cílem přiřadit oddíly na jiného člena.
Skupina spotřebitelů je vyvážena, pokud tam nejsou žádné volání poll() v této době.
Tam je další vlastnost auto.commit.interval.ms. Auto spáchat posuny zkontrolujte, zda se bude jmenovat pouze během hlasování - kontroluje, zda čas, který uplynul je větší než nakonfigurované auto commit interval a pokud je výsledek ano, offset se zavázala.
Pokud Kafka spotřebitele trvá příliš dlouho, aby proces záznamy, pak následné poll() volání je také opožděn a vyrovnává se vrátil na poslední poll() nejsou předepsány. Pokud obnovení rovnováhy se stane v této době, nové spotřebitelské klient přiřazen do tohoto oddílu zahájí zpracování zprávy znovu.
Skupina spotřebitelů obnovení rovnováhy a následné přeřazení oddílu se lze vyhnout tím, že zvýšení této hodnoty. Tím se zvýší povolený interval mezi volbám a dát více času, aby spotřebitelé zvládnout záznam(y), se vrátil z poll(). Spotřebitelé se připojit pouze k obnovení rovnováhy uvnitř volání poll, tak zvýšení maximální interval dotazování bude také zpoždění skupiny změnit rozdělení.
Tam je jeden další problém ve zvýšení maximální interval dotazování na velké hodnoty. Pokud spotřebitel zemře z jiného důvodu, to trvá déle, než nakonfigurované max.poll.interval.ms interval pro detekci selhání.
session.timeout.ms a heartbeat.interval.ms k dispozici jsou v tomto případě pro detekci celkové selhání jako dříve, jak je to možné.
Pro více informací o těchto parametrech:
Vezměte prosím na vědomí, že hodnoty nakonfigurován pro session.timeout.ms musí být v povoleném rozsahu, jak bylo nakonfigurováno v broker konfigurace podle vlastností
Jinak, následující výjimka bude vyvolána při spuštění spotřebitele klienta.
Exception in thread "main" org.apache.kafka.common.errors.InvalidSessionTimeoutException:
The session timeout is not within the range allowed by the broker
(as configured by group.min.session.timeout.ms and group.max.session.timeout.ms)
Aktualizace: Aby se zabránilo manipulaci zpráv znovu
Tam je další metoda v KafkaConsumer třídy commitAsync() vyvolat spáchat posuny provoz.
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(500));
kafkaConsumer.commitAsync();
Pro více informací o commitSync() a commitAsync(), zkontrolujte, prosím, toto vlákno
Spáchání offset ručně je akční říct, že posun byl zpracován tak, že Kafka nechce poslat dopustil záznamy pro stejnou oblast znovu. Když offsety jsou odhodláni ručně, to je důležité si uvědomit, že pokud spotřebitel zemře před zpracováním záznamů z jakéhokoli důvodu, tam je šance, tyto záznamy nebudou zpracovány znovu.