ksqlDB leaks admin client threads for terminated http2 push queries JOIN

[agavra]

Describe the bug
ksqlDB leaks threads when issued from the command line and then ^C

To Reproduce

1. Run a ksqlDB server (happens on 0.13)
2. Create a stream (CREATE STREAM a (id int key, col1 int) with (kafka_topic='a', value_format='json', partitions=1);)
3. Create another stream (CREATE STREAM c (id int key, col1 int) with (kafka_topic='c', value_format='json', partitions=1);)
4. Record the number of running threads

jstack `ps ax | grep ksql-server | awk '{print $1}' | head -1` | grep -oE '".*"' | wc -l

5. Issue a query

curl --http2 -m 240 -d '{"sql":"SELECT * FROM a JOIN c WITHIN 10 second ON a.id = c.id EMIT CHANGES;"}' http://0.0.0.0:8088/query-stream

6. Exit the query with ctrl+C
7. Check the number of running threads

Expected behavior
The number of threads should eventually be the same

Actual behaviour
The number of threads is increased by 1

Additional context
It seems that the threads which are being leaked are admin client threads:

~ ❯ jstack `ps ax | grep ksql-server | awk '{print $1}' | head -1` | grep "admin-client" | wc -l                                                                  16:55:34
      61
~ ❯ jstack `ps ax | grep ksql-server | awk '{print $1}' | head -1` | grep "admin-client" | wc -l                                                                  16:56:26
      62

Here is the stack for one such thread:

"kafka-admin-client-thread | adminclient-66" #1198 daemon prio=5 os_prio=31 tid=0x00007f9c75780000 nid=0x28607 runnable [0x000070001a71b000]
   java.lang.Thread.State: RUNNABLE
        at sun.nio.ch.KQueueArrayWrapper.kevent0(Native Method)
        at sun.nio.ch.KQueueArrayWrapper.poll(KQueueArrayWrapper.java:198)
        at sun.nio.ch.KQueueSelectorImpl.doSelect(KQueueSelectorImpl.java:117)
        at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
        - locked <0x0000000706797ce0> (a sun.nio.ch.Util$3)
        - locked <0x0000000706797cf0> (a java.util.Collections$UnmodifiableSet)
        - locked <0x0000000706797c90> (a sun.nio.ch.KQueueSelectorImpl)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
        at org.apache.kafka.common.network.Selector.select(Selector.java:869)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:465)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:561)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1319)

It also appears that we don't leak threads for just regular queries (non-join selects)... weird!

[stevenpyzhang]

With both the old and new query endpoints, a new admin client thread is created as part of checking the partition number of both source topics for the join. Whenever a request is sent to either endpoint DefaultApiSecurityContext.create(routingContext) a new service context is created.

With the old server endpoint, it looks like we're closing the admin client in a finally block here https://github.com/confluentinc/ksql/blob/master/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlServerEndpoints.java#L268

I think this is what's missing from the new server endpoint. The admin client is created and then it's never closed on the new endpoint.

[agavra]

nice find @stevenpyzhang 🥳

[apurvam]

+1, great find!

…

On Tue, Oct 27, 2020 at 8:59 AM Almog Gavra ***@***.***> wrote: nice find @stevenpyzhang <https://github.com/stevenpyzhang> 🥳 — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#6367 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAGKWDHNI5YDHA3CNJHBGVDSM3U6JANCNFSM4SGV5HSA> .