[test] Unstable test Flink22TableSourceBatchITCase.testCountPushDownForLogTable

[github-actions]

https://github.com/apache/fluss/actions/runs/28177992155/job/83459521590

[ERROR] org.apache.fluss.flink.source.Flink22TableSourceBatchITCase.testCountPushDownForLogTable(boolean)[2]  Time elapsed: 0.025 s  <<< ERROR!
org.apache.flink.table.api.ValidationException: Could not execute CREATE DATABASE: (catalogDatabase: [{}], catalogName: [testcatalog], databaseName: [defaultdb], ignoreIfExists: [false])
	at org.apache.flink.table.operations.ddl.CreateDatabaseOperation.execute(CreateDatabaseOperation.java:90)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1308)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:944)
	at org.apache.fluss.flink.source.FlinkTableSourceBatchITCase.before(FlinkTableSourceBatchITCase.java:79)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:727)
	at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
	at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:156)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptLifecycleMethod(TimeoutExtension.java:128)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptBeforeEachMethod(TimeoutExtension.java:78)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(InterceptingExecutableInvoker.java:103)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.lambda$invoke$0(InterceptingExecutableInvoker.java:93)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:92)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:86)
	at org.junit.jupiter.engine.descriptor.ClassBasedTestDescriptor.invokeMethodInExtensionContext(ClassBasedTestDescriptor.java:520)
	at org.junit.jupiter.engine.descriptor.ClassBasedTestDescriptor.lambda$synthesizeBeforeEachMethodAdapter$23(ClassBasedTestDescriptor.java:505)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeBeforeEachMethods$3(TestMethodTestDescriptor.java:174)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeBeforeMethodsOrCallbacksUntilExceptionOccurs$6(TestMethodTestDescriptor.java:202)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeBeforeMethodsOrCallbacksUntilExceptionOccurs(TestMethodTestDescriptor.java:202)
... (truncated)

[fxbing]

I took a closer look at this flaky failure. The failure seems to be caused by a race between test cleanup and auto-partition creation.

In testCountPushDownForLogTable, the first parameterized run creates a partitioned log table with auto-partition enabled. During cleanup, DROP DATABASE defaultdb CASCADE removes the database/table, but a scheduled auto-partition task may still run afterwards. When it reaches ZooKeeperClient.registerPartitionAssignmentAndMetadata(), the current code creates PartitionsZNode.path(tablePath) with creatingParentsIfNeeded(), so a stale partition creation can recreate the dropped database/table parent znodes as empty metadata paths. Then the next parameterized run fails at CREATE DATABASE defaultdb because the database path exists again.

My proposed fix is:

1. In ZooKeeperClient.registerPartitionAssignmentAndMetadata(), create only the direct /partitions node for an existing table, and stop using creatingParentsIfNeeded() for the table metadata partition parent. If the table/database has already been dropped, partition creation should fail instead of recreating parent znodes.
2. In CoordinatorEventProcessor.processCreatePartition(), skip stale CreatePartitionEvents when the table id is no longer present in the coordinator context.
3. In this count-pushdown IT, stop relying on auto-partition scheduling and add the required partitions explicitly, since this test is about count aggregate pushdown rather than auto-partition behavior.

I also plan to add regression coverage for the ZooKeeper parent recreation case and the stale coordinator partition event case, then verify the affected server tests and the original Flink 2.2 IT.

Does this direction sound reasonable? If there is any concern about changing the ZooKeeper parent creation behavior, I’d be happy to adjust the approach.