[Bug] when flinksql with group by partition some errors field occured in jobmanager.log

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

### Search before asking

- [X] I have searched in the [issues](https://github.com/apache/kyuubi/issues?q=is%3Aissue) and found no similar issues.

### Describe the bug

flink.version=1.17.1
kyuubi.version=1.8.0
hive.version=3.1.2
when run some hive sql as followings:
CREATE CATALOG bidwhive WITH ('type' = 'hive', 'hive-version' = '3.1.2', 'default-database' = 'test');
（1）select count(*) from bidwhive.test.dws_test where dt='2024-01-02' ;
+---------+
| EXPR$0 |
+---------+
| 1317 |
+---------+
It's OK. There is no errors anywhere.
（2）select dt,count(*) from bidwhive.test.dws_test where dt='2024-01-02' group by dt;

+-------------+---------+
| dt | EXPR$1 |
+-------------+---------+
| 2024-01-02 | 1317 |
+-------------+---------+

It can get correct result. But when i check jobmanager.log，I found some errors appeared repeatly as folowings.Sometimes the errors also appeared on the client terminal. I don't known whether these error will affect task runtime or not?. Can somebody help me to have a see?

### Affects Version(s)

master/1.8.0

### Kyuubi Server Log Output

_No response_

### Kyuubi Engine Log Output

```logtalk
2024-01-09 14:03:25,979 WARN org.apache.flink.streaming.api.operators.collect.CollectResultFetcher [] - An exception occurred when fetching query results
java.util.concurrent.ExecutionException: org.apache.flink.util.FlinkException: Coordinator of operator e9a3cbdf90f308bdf13b34acd6410e2b does not exist or the job vertex this operator belongs to is not initialized.
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_191]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895) ~[?:1.8.0_191]
at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.sendRequest(CollectResultFetcher.java:170) ~[flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:129) [flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106) [flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80) [flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.table.planner.connectors.CollectDynamicSink$CloseableRowIteratorWrapper.hasNext(CollectDynamicSink.java:222) [flink-table-planner_b1e58bff-c004-4dba-b7d4-fff4e8145073.jar:1.17.1]
at org.apache.flink.table.gateway.service.result.ResultStore$ResultRetrievalThread.run(ResultStore.java:155) [flink-sql-gateway-1.17.1.jar:1.17.1]
Caused by: org.apache.flink.util.FlinkException: Coordinator of operator e9a3cbdf90f308bdf13b34acd6410e2b does not exist or the job vertex this operator belongs to is not initialized.
at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.deliverCoordinationRequestToCoordinator(DefaultOperatorCoordinatorHandler.java:135) ~[flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.runtime.scheduler.SchedulerBase.deliverCoordinationRequestToCoordinator(SchedulerBase.java:1048) ~[flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.runtime.jobmaster.JobMaster.sendRequestToCoordinator(JobMaster.java:602) ~[flink-dist-1.17.1.jar:1.17.1]
at org.apache.flink.runtime.jobmaster.JobMaster.deliverCoordinationRequestToCoordinator(JobMaster.java:918) ~[flink-dist-1.17.1.jar:1.17.1]
at sun.reflect.GeneratedMethodAccessor13.invoke(Unknown Source) ~[?:?]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_191]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_191]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRpcInvocation$1(AkkaRpcActor.java:309) ~[?:?]
at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[?:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:307) ~[?:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:222) ~[?:?]
at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:84) ~[?:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:168) ~[?:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) ~[?:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) ~[?:?]
at scala.PartialFunction.applyOrElse(PartialFunction.scala:127) ~[flink-scala_2.12-1.17.1.jar:1.17.1]
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126) ~[flink-scala_2.12-1.17.1.jar:1.17.1]
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) ~[?:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175) ~[flink-scala_2.12-1.17.1.jar:1.17.1]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) ~[flink-scala_2.12-1.17.1.jar:1.17.1]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) ~[flink-scala_2.12-1.17.1.jar:1.17.1]
at akka.actor.Actor.aroundReceive(Actor.scala:537) ~[?:?]
at akka.actor.Actor.aroundReceive$(Actor.scala:535) ~[?:?]
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) ~[?:?]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:579) ~[?:?]
at akka.actor.ActorCell.invoke(ActorCell.scala:547) ~[?:?]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270) ~[?:?]
at akka.dispatch.Mailbox.run(Mailbox.scala:231) ~[?:?]
at akka.dispatch.Mailbox.exec(Mailbox.scala:243) ~[?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) ~[?:1.8.0_191]
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) ~[?:1.8.0_191]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) ~[?:1.8.0_191]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157) ~[?:1.8.0_191]
```

### Kyuubi Server Configurations

_No response_

### Kyuubi Engine Configurations

```yaml
kyuubi.engine.type=FLINK_SQL
flink.execution.target=yarn-application
```

### Additional context

_No response_

### Are you willing to submit PR?

- [ ] Yes. I would be willing to submit a PR with guidance from the Kyuubi community to fix.
- [ ] No. I cannot submit a PR at this time.

Start by reproducing the grouped and ungrouped Hive SQL queries with Flink 1.17.1, Kyuubi 1.8.0, and Hive 3.1.2, then inspect the CollectResultFetcher and ResultStore stack-trace path. Compare whether the warning affects query results or only result retrieval, and determine whether the defect belongs to Kyuubi or Flink; done means documenting the cause and a verified fix or upstream handoff.

Written by the indexing model from the issue text.