Support kotlin coroutines in consumer

[federicos-pagotic]

Hi everyone, I'm working in a project with spring-kafka and Kotlin. I have a service class with suspend functions and from the consume invoke those functions whereby the consume must be suspend function, but this not working with spring-kafka.

Service class example:

@Service
class Service() {
    suspend fun request(message: MessageBody) {
    ...
   }
}

Consume class example:

    @KafkaListener(topics = ["\${topic}"],
        groupId = "\${group}")
    @SendTo
    suspend fun consume(message: Message<MessageBody>): MessageResponse { 
         ...
         service.processMessage(message.payload)
         ...
    }

Apparently in InvocableHandlerMethod of spring-kafka not ready to work with Kotlin Coroutines, it's is correct ?

[garyrussell]

It's not really the InvocableHandlerMethod, the listener container manages the threading.

I am not that familiar with Kotlin Coroutines, but my understanding is that the listener container would have to be written in Kotlin to support them (a new listener container would be required). This will require a huge effort.

Given that project Loom (and virtual threads) is on the horizon, it is not clear that it worth the effort because supporting Loom is trivial (I already have tested it).

[federicos-pagotic]

I believe that it's not necessary write in kotlin. When in Kotlin we have a suspend function in Java it's a method with a one more parameter of type kotlin.coroutines.Continuation.

You could create a implementation of kotlin.coroutines.Continuation in Java with a Executor and then InvocableHandlerMethod send this implementation when invoke to consume.

import java.util.concurrent.Executor;
import kotlin.coroutines.Continuation;
import kotlin.coroutines.CoroutineContext;
import kotlinx.coroutines.Dispatchers;

public class JavaContinuation<T> implements Continuation<T> {

    private final Executor executor;
    private Continuation<? super T> continuation;

    public JavaContinuation(Executor executor) {
        this.executor = executor;
    }

    @Override
    public CoroutineContext getContext() {
        return Dispatchers.from(executor);
    }

    @Override
    public void resumeWith(Result<T> result) {
        continuation.resumeWith(result);
    }

    public void setContinuation(Continuation<? super T> continuation) {
        this.continuation = continuation;
    }
}

        ExecutorService executor = Executors.newSingleThreadExecutor();
        JavaContinuation<Integer> javaContinuation = new JavaContinuation<>(executor);

        CoroutineScope scope = CoroutineScopeKt.newCoroutineScope(Dispatchers.getDefault());
        scope.launch(javaContinuation.getContext(), () -> {
            Object result = consume(...);
            javaContinuation.resumeWith(Result.Companion.success(result));
        });

        executor.shutdown();

But for this you should have Kotlin coroutine dependencies. I don't know so much about the project.

There are implementations of Flux with this project ?
On the other hand, what's project Loom ?

[garyrussell]

It is not clear what that would buy you, the listener container will still have a thread running in the background and the record's offset will be committed when handed off (by default).

Exactly what are you trying to achieve by using coroutines?

There are implementations of Flux with this project

This project does not support reactive workloads.; see reactor-kafka for that.

On the other hand, what's project Loom ?

Google?

[artembilan]

There is just no listener method async return types handling in the framework yet.
We probably can do something similar we have in Spring AMQP: https://github.com/spring-projects/spring-amqp/blob/main/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java#L375-L410.
Where you can see we support CompletableFuture and Mono.
The Kotlin Coroutines Deferred result can be adapted to Mono: CoroutinesUtils.deferredToMono((Deferred<?>) source).

I don't think, Gary, that virtual threads would do anything replacing Kotlin's suspend which are in term of Java is a CompletableFuture.

[garyrussell]

We probably can do something similar we have in Spring AMQP

It is much simpler there because individual messages are acknowledged; with Kafka, all we have is a committed offset. so any implementation here would be limited to using the out of order commit deferral capability.

Use of multi-threaded consumers is discouraged because of these complications.

[federicos-pagotic]

In my first example it can be seen that use @SendTo for async return.

This project does not support reactive workloads.; see reactor-kafka for that.

Ok, I will check this project.

Google?

jajaja, you're right.

Where you can see we support CompletableFuture and Mono

It's a good perspective, but how to work @SendTo in this project for async return.

[artembilan]

Right, but see Gary's comment:

multi-threaded consumers is discouraged

I missed the fact that we need to keep an order of the processed and committed offset.
More precisely what Gary means that batch of just pulled records must be processed on the consumer thread to really preserve an order and purpose of Apache Kafka.
So, you don't have other way unless you block the target suspend function in the listener method.

[garyrussell]

Given some work going on in spring-amqp spring-projects/spring-amqp#1210 (comment) and given that the container now supports asyncAcks (where offset commits are deferred until any gaps are filled) I think we have a way forward to support Mono<?> return types (and therefore Kotlin Coroutines) in this project.

We will track it via this issue #1189