O que determina a compensação do consumidor Kafka?

170

Eu sou relativamente novo em Kafka. Eu experimentei um pouco, mas algumas coisas não estão claras para mim em relação à compensação do consumidor. Pelo que entendi até agora, quando um consumidor inicia, o deslocamento do qual ele começará a ler é determinado pela configuração auto.offset.reset(corrija-me se estiver errado).

Agora diga, por exemplo, que há 10 mensagens (compensações de 0 a 9) no tópico, e um consumidor consumiu 5 delas antes de cair (ou antes de eu matar o consumidor). Então diga que eu reinicio esse processo do consumidor. Minhas perguntas são:

  1. Se o auto.offset.resetestiver definido como smallest, sempre começará a consumir a partir do deslocamento 0?

  2. Se o auto.offset.resetestiver definido como largest, começará a consumir a partir do deslocamento 5?

  3. O comportamento em relação a esse tipo de cenário é sempre determinístico?

Por favor, não hesite em comentar se algo na minha pergunta não estiver claro. Desde já, obrigado.

Asif Iqbal
fonte

Respostas:

260

É um pouco mais complexo do que você descreveu.
A auto.offset.resetconfiguração é ativada APENAS se o seu grupo de consumidores não tiver um deslocamento válido confirmado em algum lugar (dois armazenamentos de deslocamento suportados agora são Kafka e Zookeeper) e também depende do tipo de consumidor que você usa.

Se você usa um consumidor java de alto nível, imagine os seguintes cenários:

  1. Você tem um consumidor em um grupo de consumidores group1que consumiu 5 mensagens e morreu. Na próxima vez em que você iniciar esse consumidor, ele nem usará essa auto.offset.resetconfiguração e continuará do local em que morreu, pois apenas buscará o deslocamento armazenado no armazenamento de deslocamento (Kafka ou ZK, como mencionei).

  2. Você tem mensagens em um tópico (como você descreveu) e inicia um consumidor em um novo grupo de consumidores group2. Não há deslocamento armazenado em nenhum lugar e, desta vez, a auto.offset.resetconfiguração decidirá se deve começar do início do tópico ( earliest) ou do final do tópico ( latest)

Mais uma coisa que afeta a que valor de deslocamento corresponderá earlieste latestconfigura é a política de retenção de log. Imagine que você tenha um tópico com retenção configurada para 1 hora. Você produz 5 mensagens e, uma hora depois, publica mais 5 mensagens. O latestdeslocamento ainda permanecerá o mesmo do exemplo anterior, mas earliestnão será possível 0porque o Kafka já removerá essas mensagens e, portanto, o deslocamento disponível mais antigo será 5.

Tudo mencionado acima não está relacionado SimpleConsumere, toda vez que você o executa, ele decide por onde começar usando a auto.offset.resetconfiguração.

Se você usar a versão Kafka mais velho do que 0,9, você tem que substituir earliest, latestcom smallest, largest.

serejja
fonte
3
Muito obrigado pela resposta. Então, quanto ao consumidor de alto nível, uma vez que um consumidor tenha algo comprometido (em ZK ou Kafka), auto.offset.resetisso não terá mais significado a partir de então? O único significado dessa configuração é quando não há nada comprometido (e, idealmente, isso seria na primeira inicialização do consumidor)?
Asif Iqbal 4/15
2
Exatamente como você descreveu
serejja
1
@serejja Olá - e se eu sempre tiver um consumidor por grupo e o cenário nº 1 da sua resposta ocorrer para mim? Seria o mesmo?
precisa saber é o seguinte
1
@ ha9u63ar não entendeu bem sua pergunta. Se você reiniciar o consumidor no mesmo grupo, sim, ele não será usado auto.offset.resete continuará a partir do deslocamento confirmado. Se você sempre usa grupo de consumidores diferente (como gerá-lo ao iniciar o consumidor), então o consumidor vai sempre respeitarauto.offset.reset
serejja
@serejja sim e isso não está funcionando para mim. você poderia por favor dê uma olhada neste - este é o meu problema
ha9u63ar
83

Apenas uma atualização: do Kafka 0.9 em diante, o Kafka está usando uma nova versão Java do consumidor e os nomes dos parâmetros auto.offset.reset foram alterados; Do manual:

O que fazer quando não há deslocamento inicial no Kafka ou se o deslocamento atual não existe mais no servidor (por exemplo, porque esses dados foram excluídos):

mais cedo : redefine automaticamente o deslocamento para o deslocamento mais antigo

mais recente : redefine automaticamente o deslocamento para o deslocamento mais recente

none : lança exceção ao consumidor se nenhum deslocamento anterior for encontrado para o grupo de consumidores

qualquer outra coisa: lançar exceção ao consumidor.

Eu gastei algum tempo para encontrar isso depois de verificar a resposta aceita, então achei que seria útil para a comunidade publicá-la.

Israel Zinco
fonte
9

Além disso, há offsets.retention.minutes. Se o tempo desde o último commit for> offsets.retention.minutes, auto.offset.resettambém entra em ação

Sasa Ninkovic
fonte
1
isso não parece redundante com a retenção de log? a retenção inicial deve ser baseada na retenção de logs?
Mike01010
@ mike01010 isso mesmo. Deve ser baseado na retenção de log, que é uma das soluções propostas no ticket. Prolong default value of offsets.retention.minutes to be at least twice larger than log.retention.hours. Issues.apache.org/jira/browse/KAFKA-3806
saheb
Essa resposta me assustou por um tempo, até eu verificar a documentação de offsets.retention.minutes: <b> Depois que um grupo de consumidores perde todos os seus consumidores (ou seja, fica vazio), suas compensações serão mantidas por esse período de retenção antes de serem descartadas. </b> Para autônomo consumidores (usando a atribuição manual), as compensações expirarão após o horário da última confirmação mais esse período de retenção. (Isto é para Kafka 2.3)
jumping_monkey