Add JVM shutdown hook to KafkaReceiver to close KafkaConsumer

[SujanKumarMitra]

When the JVM receives a SIGINT,
the KafkaReceiver does NOT close the underlying KafkaConsumer to prevent resource leaks.

[simonbasle]

Is that a typical arrangement @garyrussell @artembilan ?
If the kafka-client (KafkaConsumer) doesn't install such a hook and puts the responsibility on the user, doesn't it make sense that the same would be required for KafkaSender (and its #close() method)?

[artembilan]

Well, Spring for Apache Kafka project is fully rely on the ApplicationContext lifecycle.
So, whatever calls its close() would initiate the underlying KafkaConsumer.close().

Doesn't look like Spring adds such a hook by default: https://docs.spring.io/spring-framework/docs/current/reference/html/core.html#beans-factory-shutdown.
But at least there is such a API for end-user consideration.

Looks like Spring Boot has its own opinion: https://github.com/spring-projects/spring-boot/blob/main/spring-boot-project/spring-boot/src/main/java/org/springframework/boot/SpringApplication.java#L203.

But anyway. If we expose something like registerShutdownHook() on the KafkaReceiver and KafkaSender, that should be enough for end-user requirements.

[SujanKumarMitra]

I am using Reactor Kafka in Spring Boot Webflux environment.

So, I am expecting that spring-managed KafkaReceiver beans should be closed on SIGINT. Like it happens for JDBC or MongoDB connections.

But, Spring only closes KafkaSender connections. Maybe if KafkaReceiver implements the Closeable interface or close(), spring's shutdown hook might kick in.

[artembilan]

OK. I see. Right, the KafkaReceiver (neither its impl) has a close() API, unlike KafkaSender does.

At the same time I see that ConsumerHandler has one, but does not delegate to its Consumer.close().

[garyrussell]

ConsumerHandler invokes consumerEventLoop.stop() which schedules a CloseEvent which closes the consumer.

It looks to me like the handler.close() will be invoked when the receiver.receive() flux completes; it is passed into the usingWhen as the asyncCleanUp argument.

What am I missing? Isn't it the application's responsibility to complete the flux?

[SujanKumarMitra]

I have a question. When exactly does the KafkaReceiver.receive() completes? I think, it runs indefinitely unless someone uses the Disposable.dispose() method returned by Flux.subscribe().

My proposal is,
expose a close() method in KafkaReceiver which will complete the KafkaReceiver.receive() flux and close the KafkaConsumer as well.

Also, KafkaSender has a close() but it doesn't extend the Closeable or AutoCloseable. It would be better if it extends the AutoCloseable so that it can be used with try-with-resources

[garyrussell]

Your application bean could implement AutoCloseable and call dispose().

Also, KafkaSender has a close() but it doesn't extend the Closeable or AutoCloseable. It would be better if it extends the AutoCloseable so that it can be used with try-with-resources

Agreed.

[SujanKumarMitra]

Your application bean could implement AutoCloseable and call dispose().

Yes, currently my component class implements the DisposableBean and, in the destroy() method, I am calling the dispose() method.
But the problem is, the subscribers of the KafkaReceiver.receive() receive an error signal of CancellationException when dispose() is called. That's why I am proposing this:

My proposal is,
expose a close() method in KafkaReceiver which will complete the KafkaReceiver.receive() flux and close the KafkaConsumer as well.

[simonbasle]

Also, KafkaSender has a close() but it doesn't extend the Closeable or AutoCloseable. It would be better if it extends the AutoCloseable so that it can be used with try-with-resources

Isn't there an inherent risk here, because the class is reactive and thus non-blocking?
A try-with-resource block would only make sense if you block inside, which is generally discouraged.
Worse, if you don't block but still use the KafkaSender inside a twr, it will get closed too early.

I would be vary of such a change.

[simonbasle]

(exposing a close() method where it is missing makes perfect sense though)