ksqlDB happens a Thread blocked issue when start 3 pods in the Kubernetes

**Describe the bug**

When I use the Rest API to query a table from the ksqlDB, I always can get the "io.vertx.core.VertxException: Thread blocked" issue from the ksqlDB server pod. especially when I start more than one ksqlDB pod (The number of pods I tested was 3), but if only starting one pod cannot hit the Thread blocked issue.
The issue was very easy to reproduce, we have 3-4 people do the test at the same time, we connect to the ksqlDB server, and queried data from the same table after a while the "Thread blocked" issue will happen (10-40 minutes), once the issue happens the pod will never return any data until restart it. I find a very similar issue at #5600 and from the comments looks like the issue has already been fixed but I still can reproduce the issue, so I decided to report it again.

And I find an easy way can quickly reproduce the issue please see the following steps.

**To Reproduce**

1. The version of KSQL.

I have tested **0.28.2** and **0.27.1** both of them have the same issue.

2. Sample source data.

I have sent the JSON format data to Kafka by topic "cd-sim-non-periodic"

Java code for sending the data is:

` KafkaProducer.send(new ProducerRecord("cd-sim-non-periodic", uuid.toString(), json)); `

sample of the json

```
{
"ts": 1681365028455,
"vin": "0000000003abcdefg",
"num": "3505646",
"speed": "348.7543574578107"
}
```

3. Any SQL statements you ran

Create the table in the ksqlDB script is:

```
CREATE source table IF NOT EXISTS test_table_1 (`id` varchar primary key, `vin` varchar, `ts` bigint, num varchar, speed varchar) with (KAFKA_TOPIC='cd-sim-non-periodic'
>, VALUE_FORMAT='JSON');
```

Create the table for the query:

` create table test_table_query as select * from test_table;`

4. reproduce the issue

Use Postman as the test tool.

a. set the header like this:

b. set the body like this :

```
{
"ksql": "select * from test_table_query limit 190000;"
}
```

Click the Send button again and again. Don't wait for the result. after about 30 times the ksqlDB server will no respond.

**Expected behavior**

The server side should have a way to handle the Thread blocked issue, when the issue happens the ksqlDB server should keep working as normal instead of having no response.

**Actual behavior**

The ksqlDB server has no response until restart.

1. KSQL logs

Over and over again get the following logs

```
[2023-04-13 06:17:01,034] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,049] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,134] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,149] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,234] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,249] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,334] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,349] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,434] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,449] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,534] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,549] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,634] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,649] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,734] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,749] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,834] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,849] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,934] INFO 10.244.0.36 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
[2023-04-13 06:17:01,949] INFO 10.244.4.14 - - [Thu, 13 Apr 2023 06:17:01 GMT] "POST /heartbeat HTTP/1.1" 200 13 "-" "-" 126 (io.confluent.ksql.api.server.LoggingHandler:144)
Apr 13, 2023 6:17:02 AM io.vertx.core.impl.BlockedThreadChecker
WARNING: Thread Thread[vert.x-eventloop-thread-4,5,main]=Thread[vert.x-eventloop-thread-4,5,main] has been blocked for 5121681 ms, time limit is 2000 ms
io.vertx.core.VertxException: Thread blocked
at app//io.confluent.ksql.query.PullQueryQueue.acceptRow(PullQueryQueue.java:225)
at app//io.confluent.ksql.query.PullQueryQueue.acceptRows(PullQueryQueue.java:196)
at app//io.confluent.ksql.execution.pull.HARouting.lambda$streamedRowsHandler$15(HARouting.java:465)
at app//io.confluent.ksql.execution.pull.HARouting$$Lambda$1841/0x000000080093ec40.accept(Unknown Source)
at app//io.confluent.ksql.rest.client.KsqlTarget.lambda$null$11(KsqlTarget.java:347)
at app//io.confluent.ksql.rest.client.KsqlTarget$$Lambda$1887/0x0000000800954040.handle(Unknown Source)
at app//io.vertx.core.parsetools.impl.RecordParserImpl.handleParsing(RecordParserImpl.java:214)
at app//io.vertx.core.parsetools.impl.RecordParserImpl.handle(RecordParserImpl.java:285)
at app//io.vertx.core.parsetools.impl.RecordParserImpl.handle(RecordParserImpl.java:27)
at app//io.vertx.core.http.impl.HttpClientResponseImpl.handleChunk(HttpClientResponseImpl.java:232)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.lambda$beginResponse$0(Http1xClientConnection.java:486)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl$$Lambda$1334/0x0000000800858c40.handle(Unknown Source)
at app//io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:237)
at app//io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:127)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.handleChunk(Http1xClientConnection.java:322)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.access$1900(Http1xClientConnection.java:242)
at app//io.vertx.core.http.impl.Http1xClientConnection.handleResponseChunk(Http1xClientConnection.java:631)
at app//io.vertx.core.http.impl.Http1xClientConnection.handleHttpMessage(Http1xClientConnection.java:601)
at app//io.vertx.core.http.impl.Http1xClientConnection.handleMessage(Http1xClientConnection.java:577)
at app//io.vertx.core.net.impl.VertxHandler$$Lambda$1308/0x000000080083d440.handle(Unknown Source)
at app//io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366)
at app//io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:43)
at app//io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:229)
at app//io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:164)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at app//io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at app//io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at app//io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
at app//io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
at app//io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at app//io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at app//io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at app//io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at app//io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at app//io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
at app//io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
at app//io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
at app//io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
at app//io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
at app//io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at app//io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base@11.0.13/java.lang.Thread.run(Thread.java:829)

Apr 13, 2023 6:17:02 AM io.vertx.core.impl.BlockedThreadChecker
WARNING: Thread Thread[vert.x-eventloop-thread-3,5,main]=Thread[vert.x-eventloop-thread-3,5,main] has been blocked for 5121682 ms, time limit is 2000 ms
io.vertx.core.VertxException: Thread blocked
at java.base@11.0.13/jdk.internal.misc.Unsafe.park(Native Method)
at java.base@11.0.13/java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:234)
at java.base@11.0.13/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2123)
at java.base@11.0.13/java.util.concurrent.ArrayBlockingQueue.offer(ArrayBlockingQueue.java:393)
at app//io.confluent.ksql.query.PullQueryQueue.doAcceptRow(PullQueryQueue.java:256)
at app//io.confluent.ksql.query.PullQueryQueue.acceptRow(PullQueryQueue.java:236)
at app//io.confluent.ksql.query.PullQueryQueue.acceptRows(PullQueryQueue.java:196)
at app//io.confluent.ksql.execution.pull.HARouting.lambda$streamedRowsHandler$15(HARouting.java:465)
at app//io.confluent.ksql.execution.pull.HARouting$$Lambda$1841/0x000000080093ec40.accept(Unknown Source)
at app//io.confluent.ksql.rest.client.KsqlTarget.lambda$null$11(KsqlTarget.java:347)
at app//io.confluent.ksql.rest.client.KsqlTarget$$Lambda$1887/0x0000000800954040.handle(Unknown Source)
at app//io.vertx.core.parsetools.impl.RecordParserImpl.handleParsing(RecordParserImpl.java:214)
at app//io.vertx.core.parsetools.impl.RecordParserImpl.handle(RecordParserImpl.java:285)
at app//io.vertx.core.parsetools.impl.RecordParserImpl.handle(RecordParserImpl.java:27)
at app//io.vertx.core.http.impl.HttpClientResponseImpl.handleChunk(HttpClientResponseImpl.java:232)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.lambda$beginResponse$0(Http1xClientConnection.java:486)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl$$Lambda$1334/0x0000000800858c40.handle(Unknown Source)
at app//io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:237)
at app//io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:127)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.handleChunk(Http1xClientConnection.java:322)
at app//io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.access$1900(Http1xClientConnection.java:242)
at app//io.vertx.core.http.impl.Http1xClientConnection.handleResponseChunk(Http1xClientConnection.java:631)
at app//io.vertx.core.http.impl.Http1xClientConnection.handleHttpMessage(Http1xClientConnection.java:601)
at app//io.vertx.core.http.impl.Http1xClientConnection.handleMessage(Http1xClientConnection.java:577)
at app//io.vertx.core.net.impl.VertxHandler$$Lambda$1308/0x000000080083d440.handle(Unknown Source)
at app//io.vertx.core.impl.ContextImpl.executeTask(ContextImpl.java:366)
at app//io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:43)
at app//io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:229)
at app//io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:164)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at app//io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at app//io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
at app//io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
at app//io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
at app//io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
at app//io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
at app//io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at app//io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at app//io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at app//io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at app//io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at app//io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at app//io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
at app//io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
at app//io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
at app//io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
at app//io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
at app//io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at app//io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base@11.0.13/java.lang.Thread.run(Thread.java:829)

```

**Additional context**

my configuration of the ksqlDB pod

```
KSQL_BOOTSTRAP_SERVERS : ev-kafka-bootstrap.kafka:9092
KSQL_KSQL_ADVERTISED_LISTENER : http://$(POD_NAME).ksqldb-server-headless.$(POD_NAMESPACE).svc.cluster.local:8099
KSQL_KSQL_HEARTBEAT_ENABLE : true
KSQL_KSQL_INTERNAL_LISTENER : http://0.0.0.0:8099
KSQL_KSQL_LAG_REPORTING_ENABLE : true
KSQL_KSQL_QUERY_PULL_ENABLE_STANDBY_READS : true
KSQL_KSQL_QUERY_PULL_METRICS_ENABLE : true
KSQL_KSQL_SERVICE_ID : evcar_service_
KSQL_KSQL_STREAMS_NUM_STANDBY_REPLICAS : 1
KSQL_LISTENERS : http://0.0.0.0:8088
KSQL_STREAMS_AUTO_OFFSET_RESET : earliest
```

Start with the stack-trace locations PullQueryQueue.java, HARouting.java, and KsqlTarget.java, then reproduce the failure using three ksqlDB pods and repeated REST queries for the 190000-row table. Compare behavior with one pod and versions 0.28.2 and 0.27.1; done means the server continues returning data without the Vert.x blocked-thread condition or requiring a restart.

Written by the indexing model from the issue text.