FlinkTestBase.waitUntilPartitions is not stable

### Search before asking

- [x] I searched in the [issues](https://github.com/alibaba/fluss/issues) and found nothing similar.

### Fluss version

main (development)

### Please describe the bug 🐞

Some cases are failed on the unstable method `FlinkTestBase.waitUntilPartitions`

https://github.com/alibaba/fluss/actions/runs/13803475787/job/38635627295?pr=586
```
FlinkTableSourceITCase.testLookupPartitionedTable:846->prepareDimTableAndSourceTable:1005->FlinkTestBase.waitUntilPartitions:209->FlinkTestBase.waitUntilPartitions:221 expect 2 table partition has not been created
```

https://github.com/alibaba/fluss/actions/runs/13803475799/job/38635627330?pr=586
```
FlinkTableSourceFailOverITCase.testRestore:154 expect 2 table partition has not been created
```

I downloaded the artifact logs, and found some Suspicious logs, but still not sure the root cause:

```
21:20:49,876 [coordinator-event-thread] ERROR com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager$CoordinatorEventThread [] - Uncaught error processing event DropPartitionEvent{tableId=0, partitionId=0, partitionName='4000}.
java.lang.NullPointerException: null
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.processDropPartition(CoordinatorEventProcessor.java:562) ~[fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.process(CoordinatorEventProcessor.java:447) ~[fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager$CoordinatorEventThread.doWork(CoordinatorEventManager.java:99) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.utils.concurrent.ShutdownableThread.run(ShutdownableThread.java:95) [fluss-client-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
21:20:49,879 [ForkJoinPool-2-worker-1] INFO com.alibaba.fluss.server.zk.ZooKeeperClient [] - Deleted table fluss.test_partitioned.
21:20:49,879 [coordinator-event-thread] ERROR com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager$CoordinatorEventThread [] - Uncaught error processing event DropTableEvent{tableId=0, isAutoPartitionTable=true}.
java.lang.NullPointerException: null
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.processDropTable(CoordinatorEventProcessor.java:541) ~[fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.process(CoordinatorEventProcessor.java:445) ~[fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager$CoordinatorEventThread.doWork(CoordinatorEventManager.java:99) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.utils.concurrent.ShutdownableThread.run(ShutdownableThread.java:95) [fluss-client-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
21:20:49,883 [ForkJoinPool-2-worker-1] INFO com.alibaba.fluss.server.zk.ZooKeeperClient [] - Deleted table fluss.result_table.
21:20:49,883 [coordinator-event-thread] ERROR com.alibaba.fluss.server.coordinator.statemachine.TableBucketStateMachine [] - Failed to move table buckets [TableBucket{tableId=1, bucket=0}] to state OfflineBucket.
java.lang.IllegalStateException: The NotifyLeaderAndIsr batch request from coordinator to tablet server is not empty while creating a new one. Some NotifyLeaderAndIsr request in {1={}} might be lost.
at com.alibaba.fluss.server.coordinator.CoordinatorRequestBatch.newBatch(CoordinatorRequestBatch.java:94) ~[fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.statemachine.TableBucketStateMachine.handleStateChange(TableBucketStateMachine.java:112) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.TableManager.onDeleteTable(TableManager.java:163) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.processDropTable(CoordinatorEventProcessor.java:549) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.process(CoordinatorEventProcessor.java:445) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager$CoordinatorEventThread.doWork(CoordinatorEventManager.java:99) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.utils.concurrent.ShutdownableThread.run(ShutdownableThread.java:95) [fluss-client-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
21:20:49,883 [coordinator-event-thread] ERROR com.alibaba.fluss.server.coordinator.statemachine.TableBucketStateMachine [] - Failed to move table buckets [TableBucket{tableId=1, bucket=0}] to state NonExistentBucket.
java.lang.IllegalStateException: The NotifyLeaderAndIsr batch request from coordinator to tablet server is not empty while creating a new one. Some NotifyLeaderAndIsr request in {1={}} might be lost.
at com.alibaba.fluss.server.coordinator.CoordinatorRequestBatch.newBatch(CoordinatorRequestBatch.java:94) ~[fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.statemachine.TableBucketStateMachine.handleStateChange(TableBucketStateMachine.java:112) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.TableManager.onDeleteTable(TableManager.java:164) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.processDropTable(CoordinatorEventProcessor.java:549) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.CoordinatorEventProcessor.process(CoordinatorEventProcessor.java:445) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.server.coordinator.event.CoordinatorEventManager$CoordinatorEventThread.doWork(CoordinatorEventManager.java:99) [fluss-server-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
at com.alibaba.fluss.utils.concurrent.ShutdownableThread.run(ShutdownableThread.java:95) [fluss-client-0.7-SNAPSHOT.jar:0.7-SNAPSHOT]
```

Another log

```
21:32:32,224 [coordinator-event-thread] INFO com.alibaba.fluss.server.coordinator.TableManager [] - Deletion of partition TablePartition{tableId=47, partitionId=18} successfully completed.
21:32:32,225 [fluss-netty-server-worker-thread-2] INFO com.alibaba.fluss.server.zk.ZooKeeperClient [] - Registered new schema version 1 for table defaultdb.lookup_test_enable_cache_async_pk_id_451088596.
21:32:32,226 [fluss-netty-server-worker-thread-2] INFO com.alibaba.fluss.server.zk.ZooKeeperClient [] - Registered table lookup_test_enable_cache_async_pk_id_451088596 for database defaultdb
21:32:32,241 [fluss-netty-client(NIO)-82-1] WARN com.alibaba.fluss.client.lookup.LookupSender [] - Get error lookup response on table bucket TableBucket{tableId=47, partitionId=18, bucket=0}, fail. Error: UNKNOWN_TABLE_OR_BUCKET_EXCEPTION. Error Message: Unknown table or bucket: TableBucket{tableId=47, partitionId=18, bucket=0}
21:32:32,241 [Legacy Source Thread - Source: Collection Source -> *anonymous_datastream_source$85*[201] -> Calc[202] -> LookupJoin[203] -> Calc[204] -> Sink: Collect table sink (1/1)#0] ERROR com.alibaba.fluss.connector.flink.source.lookup.FlinkLookupFunction [] - Fluss lookup error, retry times = 1
java.util.concurrent.ExecutionException: com.alibaba.fluss.exception.UnknownTableOrBucketException: Unknown table or bucket: TableBucket{tableId=47, partitionId=18, bucket=0}
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_432]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908) ~[?:1.8.0_432]
at com.alibaba.fluss.connector.flink.source.lookup.FlinkLookupFunction.lookup(FlinkLookupFunction.java:128) ~[classes/:?]
at org.apache.flink.table.functions.LookupFunction.eval(LookupFunction.java:52) ~[flink-table-common-1.19.2.jar:1.19.2]
at LookupFunction$5.flatMap(Unknown Source) ~[?:?]
at org.apache.flink.table.runtime.operators.join.lookup.LookupJoinRunner.doFetch(LookupJoinRunner.java:110) ~[flink-table-runtime-1.19.2.jar:1.19.2]
at org.apache.flink.table.runtime.operators.join.lookup.LookupJoinRunner.processElement(LookupJoinRunner.java:92) ~[flink-table-runtime-1.19.2.jar:1.19.2]
at org.apache.flink.table.runtime.operators.join.lookup.LookupJoinRunner.processElement(LookupJoinRunner.java:35) ~[flink-table-runtime-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at StreamExecCalc$4.processElement(Unknown Source) ~[?:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) ~[flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.table.runtime.operators.source.InputConversionOperator.processElement(InputConversionOperator.java:128) ~[flink-table-runtime-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollect(StreamSourceContexts.java:425) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collect(StreamSourceContexts.java:520) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.operators.StreamSourceContexts$SwitchingOnClose.collect(StreamSourceContexts.java:110) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.functions.source.FromElementsFunction.run(FromElementsFunction.java:237) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:113) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:71) [flink-streaming-java-1.19.2.jar:1.19.2]
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:338) [flink-streaming-java-1.19.2.jar:1.19.2]
Caused by: com.alibaba.fluss.exception.UnknownTableOrBucketException: Unknown table or bucket: TableBucket{tableId=47, partitionId=18, bucket=0}
21:32:32,703 [ Thread-467] INFO com.alibaba.fluss.rpc.netty.client.NettyClient [] - Netty client was shutdown successfully.
```

### Solution

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Start by reading FlinkTestBase.waitUntilPartitions, which the report identifies as the failing helper, and compare the failures in FlinkTableSourceITCase.testLookupPartitionedTable and FlinkTableSourceFailOverITCase.testRestore. Inspect the coordinator partition/table deletion errors in the supplied logs alongside the partition-creation wait. Done means these tests reliably observe the expected two partitions without the reported failure.

Written by the indexing model from the issue text.