fix

Author: JingsongLi

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriteOperatorCoordinator.java

@@ -19,7 +19,10 @@
 package org.apache.paimon.flink.sink.coordinator;
 
 import org.apache.paimon.flink.sink.TableWriteOperator;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SegmentsCache;
 
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
@@ -32,6 +35,7 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ThreadPoolExecutor;
 
+import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_COORDINATOR_CACHE_MEMORY;
 import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
 
 /**
@@ -52,6 +56,10 @@ public WriteOperatorCoordinator(FileStoreTable table) {
     @Override
     public void start() throws Exception {
         executor = createCachedThreadPool(1, "WriteCoordinator");
+        MemorySize cacheMemory =
+                table.coreOptions().toConfiguration().get(SINK_WRITER_COORDINATOR_CACHE_MEMORY);
+        SegmentsCache<Path> manifestCache = SegmentsCache.create(cacheMemory, Long.MAX_VALUE);
+        table.setManifestCache(manifestCache);
         coordinator = new TableWriteCoordinator(table);
     }