Open 62mkv opened 1 year ago
Looks like this is this your SO question, too: https://stackoverflow.com/questions/76015312/how-to-properly-deal-with-zombie-kafka-consumers-in-reactive-spring-boot-appli
this was originally asked on SO here: https://stackoverflow.com/questions/76015312/how-to-properly-deal-with-zombie-kafka-consumers-in-reactive-spring-boot-appli but as I read the code afterwards, it feels more and more like a bug :(
I see. Any thoughts about the possible fix? Or share, please, with us what part of the project code you think is producing such a bug?
Sent from Mail.ru app for Android Friday, 14 April 2023, 06:52pm +03:00 from Artem Bilan @.*** :
I see. Any thoughts about the possible fix? Or share, please, with us what part of the project code you think is producing such a bug? ā Reply to this email directly, view it on GitHub , or unsubscribe . You are receiving this because you authored the thread. Message ID: @ github . com>
You need to add retry (and possibly repeat) to the pipeline: https://projectreactor.io/docs/kafka/release/reference/#_error_handling_2
@garyrussell I see. Thanks so much for your help!
I will give it a try and close the ticket when I can confirm it's no longer manifesting.
May I also ask about a different thing: how useful would it be, to have a partition revoke handler commit offsets?
var receiverOptions = ReceiverOptions.create(getConsumerProperties())
.commitInterval(DEFAULT_COMMIT_INTERVAL)
.addAssignListener(this::handlePartitionsAssignment)
.addRevokeListener(this::handlePartitionsRevoking)
.subscription(activeDestinations);
...
@SneakyThrows(IllegalAccessException.class)
private void handlePartitionsRevoking(Collection<ReceiverPartition> revokedPartitions) {
var consumer = (Consumer<Object, Object>) FieldUtils.readField(kafkaReceiver.consumerHandler(), "consumer", true);
try {
consumer.commitSync(latestAcks, Duration.ofMillis(1000));
} catch (Exception e) {
log.warn("Ignored error on partition revoke", e);
}
}
It seems that the documentation says that downstream consumer should not be concerned with this:
All acknowledged offsets are committed when partitions are revoked during rebalance and when the receive Flux is terminated
So this quoted code in the handler, presented above, is some legacy code and could/should be removed? Or it might be needed still? We have WakeupException
inside this handler occasionally so I thought I'd ask some experts..
I don't know what latestAcks
is there but, indeed, the CommitEvent
is run when partitions are revoked so any acknowledged records will be committed. See ConsumerEventLoop.onPartitionsRevoked
:
Expected Behavior
Actual Behavior
We have a Reactive Spring Boot application that employs "reactor-kafka" for Kafka consumers and producers.
we use 1
KafkaReceiver
per topic, that is subscribed to and kept in a Spring bean field.I observe that sometimes, some or all of the underlying
Consumer
-s just stop with an error message as follows:(this is the last message in the log thus far; the application lives happily for a day already, after all 11 of consumers have stuck in this limbo; topic is consumed by other pods)
Regardless of what the error says, should not consumer still be restarted by the library/Kafka internals? Or should it be application author's responsibility to somehow track this state and react accordingly (for example, by implementing liveness health check around this somehow)?
Steps to Reproduce
Possible Solution
Your Environment
netty
, ...): reactor-kafka: 1.3.17java -version
): 11.0.16.1uname -a
): Linux x64