Spring Kafka not working properly with Java 21 Virtual Threads, listeners are tied to a platform thread forever

[Moscagus]

jstack.txt
jcmd.txt
Spring Boot 3.2.3

Virtual Kafka listeners are tied to a platform thread forever.

The problem I see is that validating with jcmd and jstack the listeners are tied to a platform thread forever.
Therefore it does not make effective use of virtual threads.
Without messages in Kafka and with a concurrency of 20, I have 20 platform threads carrying the virtual threads.

I have 3 cores. Therefore I have to have only 3 ForkJoinPool-1-worker platform threads waiting.
But in reality I have 3 waiting and another 20 carrying the listeners.

[oracle@vtuerto kafka-2602]$ cat jstack.txt | grep -A 1 ForkJoinPool-1-worker | grep WAITING
java.lang.Thread.State: TIMED_WAITING (parking)
java.lang.Thread.State: WAITING (parking)
java.lang.Thread.State: WAITING (parking)
[oracle@vtuerto kafka-2602]$ cat jstack.txt | grep -A 1 ForkJoinPool-1-worker | grep Carrying
Carrying virtual thread #38
Carrying virtual thread #73
Carrying virtual thread #47
Carrying virtual thread #76
Carrying virtual thread #80
Carrying virtual thread #61
Carrying virtual thread #43
Carrying virtual thread #92
Carrying virtual thread #45
Carrying virtual thread #64
Carrying virtual thread #84
Carrying virtual thread #88
Carrying virtual thread #69
Carrying virtual thread #96
Carrying virtual thread #100
Carrying virtual thread #104
Carrying virtual thread #108
Carrying virtual thread #112
Carrying virtual thread #116
Carrying virtual thread #51

[artembilan]

Any chances to confirm that this kind of dead-lock is indeed caused by Spring for Apache Kafka?
How about to make something similar based on virtual threads, but just with plain Apache Kafka client?
See more info in the similar issue: #3013

The virtual threads is totally Java feature, not Spring, so making them working, the whole call stack should follow specific recommendation.
The synchronized block is one of those which causes virtual threads to be pinned to platform one.

Thanks for understanding!

[Moscagus]

When checking the stacks in "jcmd.txt", I found a place where the Apache Kafka client due to a "synchronized" causes virtual thread to be pinned to plataform one.
It is the heartbeat thread for each spring kafka listener (concurrency).
I give as an example one:

       "tid": "51",
       "name": "kafka-coordinator-heartbeat-thread | Group_topic_1",
       "stack": [
          "java.base\/jdk.internal.misc.Unsafe.park(Native Method)",
          "java.base\/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)",
          "java.base\/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:754)",
          "java.base\/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:990)",
          "java.base\/java.util.concurrent.locks.ReentrantLock$Sync.lock(ReentrantLock.java:153)",
          "java.base\/java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:322)",
          "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:264)",
          "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:321)",
          "org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:1460)"
       ]

Looking at the code on github for "AbstractCoordinator$HeartbeatThread.run" you can clearly see the synchornized.
https://github.com/apache/kafka/blob/3.6/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java#L1443

    @Override
    public void run() {
        try {
            log.debug("Heartbeat thread started");
            while (true) {
                synchronized (AbstractCoordinator.this) {

It would be good if you can contact the Kafka people about this issue, otherwise it doesn't make any sense that the documentation says that Spring for Apache Kafka supports virtual threads.

[artembilan]

Thank you for the report!
Yes, we will try to find some contact with Apache community.

Would you mind to do the same via their public channels?

[artembilan]

Well, there is not too much we can do.
Looks like Kafka client creates those unconditionally:

    private synchronized void startHeartbeatThreadIfNeeded() {
        if (heartbeatThread == null) {
            heartbeatThread = new HeartbeatThread();
            heartbeatThread.start();
        }
    }

Which really are not configurable to be virtual.

This one is called from the stack of the KafkaConsumer.poll().
So, yeah: every single KafkaConsumer instance creates that HeartbeatThread for itself.
The Kafka client needs to look into making that one as virtual.

For now it is apparent that concurrency for Kafka listener must not be more than platform threads.
Or even a bit less since we still need to have some threads to run those KafkaConsumer.poll() calls...

[sobychacko]

I created an issue in the Apache Kafka jira backlog to track this: https://issues.apache.org/jira/browse/KAFKA-16309.

[Moscagus]

exactly as you mention:
https://github.com/apache/kafka/blob/3.6/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L734
the problem begins

Thanks

[Moscagus]

I continued investigating and in the end the problem is not in the Kafka consumer.
The Kafka Consumer uses the Java core "sun.nio.ch.SelectorImpl". This class uses "synchronized".
Rather than opening a case against Apache Kafka, we have to wait for a future version of Java to resolve "synchronized".

https://github.com/openjdk/jdk/blob/master/src/java.base/share/classes/sun/nio/ch/SelectorImpl.java#L123

Until this is resolved, Spring Kafka should not enable by default "SimpleAsyncTaskExecutor" as its "ListenerTaskExecutor"

[artembilan]

Well. It is still can be enabled by default, just the concurrency must not exceed the number of platform threads.
Is there a ticket in JDK we can follow about that SelectorImpl?

[Moscagus]

Although it does not seem correct to me to use a feature when it has such a clear limitation, at the very least you would have to document this limit

[sobychacko]

We can certainly document it. In addition, I don't think it is entirely a JDK issue; as @artembilan pointed out earlier in the comments, Kafka consumer also creates the HeartbeatThread in a synchronized block. So that needs to be addressed for sure. Also, did you verify if there is a corresponding JDK issue for addressing the SelectorImpl issue?