Falha no auto-commit assíncrono de offsets
Tenho uma pergunta sobre o mecanismo de confirmação automática do Kafka. Estou usando Spring-Kafka com confirmação automática habilitada. Como uma experiência, desconectei minha conexão de consumidor com Kafka por 30 segundos enquanto o sistema estava ocioso (nenhuma mensagem nova no tópico, nenhuma mensagem sendo processada). Depois de reconectar, recebi algumas mensagens como:
Asynchronous auto-commit of offsets {cs-1915-2553221872080030-0=OffsetAndMetadata{offset=19, leaderEpoch=0, metadata=''}} failed: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
Em primeiro lugar, não entendo o que há para comprometer? O sistema estava ocioso (todas as mensagens anteriores já foram confirmadas). Segundo, o tempo de desconexão foi de 30 segundos, muito menos do que 5 minutos (300000 ms) max.poll.interval.ms Terceiro, em uma falha não controlada do Kafka recebi pelo menos 30K mensagens desse tipo, que foi resolvido reiniciando o processo. Por que isso está acontecendo?
Estou listando aqui minha configuração de consumidor:
allow.auto.create.topics = true
auto.commit.interval.ms = 100
auto.offset.reset = latest
bootstrap.servers = [kafka1-eu.dev.com:9094, kafka2-eu.dev.com:9094, kafka3-eu.dev.com:9094]
check.crcs = true
client.dns.lookup = default
client.id =
client.rack =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = true
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = feature-cs-1915-2553221872080030
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
isolation.level = read_uncommitted
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = SSL
send.buffer.bytes = 131072
session.timeout.ms = 15000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
ssl.key.password = [hidden]
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = /home/me/feature-2553221872080030.keystore
ssl.keystore.password = [hidden]
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = /home/me/feature-2553221872080030.truststore
ssl.truststore.password = [hidden]
ssl.truststore.type = JKS
value.deserializer = class org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
Respostas
Em primeiro lugar, não entendo o que há para comprometer?
Você está certo, não há nada novo para confirmar se nenhum novo dado estiver fluindo. No entanto, tendo auto.commit ativado e seu consumidor ainda está em execução (mesmo sem ser capaz de se conectar ao corretor), o método de votação ainda é responsável pelas seguintes etapas:
- Obter mensagens de partições atribuídas
- Atribuição de partição de gatilho (se necessário)
- Confirmar compensações se a confirmação de compensação automática estiver habilitada
Junto com seu intervalo de 100ms (consulte auto.commit.intervals), o consumidor ainda tenta confirmar de forma assíncrona a posição de deslocamento (sem alteração) do consumidor.
Em segundo lugar, o tempo de desconexão foi de 30 segundos, muito menos que os 5 minutos (300.000 ms) max.poll.interval.ms
Não é o max.poll.interval que está causando o rebalanceamento, mas sim a combinação de sua heartbeat.interval.msconfiguração e o session.timeout.ms. Seu consumidor envia pulsações de thread em segundo plano com base na configuração do intervalo, no seu caso, 3 segundos. Se nenhuma pulsação for recebida pelo corretor antes do término do tempo limite da sessão (no seu caso, 15 segundos), o corretor removerá este cliente do grupo e iniciará um rebalanceamento.
Uma descrição mais detalhada da configuração que mencionei é fornecida na documentação do Kafka em Configurações do consumidor
Terceiro, em uma falha não controlada do Kafka, recebi pelo menos 30 mil mensagens desse tipo, que foram resolvidas reiniciando o processo. Por que isso está acontecendo?
Isso parece ser uma combinação das duas primeiras perguntas, em que os batimentos cardíacos não podem ser enviados e ainda assim o consumidor está tentando se comprometer por meio do método de pesquisa continuamente denominado.
Como @GaryRussell mencionou em seu comentário, eu teria o cuidado de usar auto.commit.enablede, em vez disso, assumir o controle sobre o gerenciamento de compensação para você mesmo.