[FLINK-38784][pipeline-connector][paimon] Bump Paimon version to 1.3.1 (#4189)

Signed-off-by: Pei Yu <[email protected]>

Author: sd4324530

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml

@@ -29,7 +29,7 @@ limitations under the License.
     <artifactId>flink-cdc-pipeline-connector-paimon</artifactId>
 
     <properties>
-        <paimon.version>1.2.0</paimon.version>
+        <paimon.version>1.3.1</paimon.version>
         <hadoop.version>2.8.5</hadoop.version>
         <hive.version>2.3.9</hive.version>
         <mockito.version>3.4.6</mockito.version>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/StoreSinkWriteImpl.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.memory.MemoryPoolFactory;
 import org.apache.paimon.memory.MemorySegmentPool;
 import org.apache.paimon.operation.FileStoreWrite;
+import org.apache.paimon.operation.WriteRestore;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.SinkRecord;
@@ -118,7 +119,7 @@ private TableWriteImpl<?> newTableWrite(FileStoreTable table) {
                 "memoryPool and memoryPoolFactory cannot be set at the same time.");
 
         TableWriteImpl<?> tableWrite =
-                table.newWrite(commitUser, (part, bucket) -> true, null)
+                table.newWrite(commitUser)
                         .withIOManager(paimonIOManager)
                         .withIgnorePreviousFiles(ignorePreviousFiles);
 
@@ -129,19 +130,25 @@ private TableWriteImpl<?> newTableWrite(FileStoreTable table) {
         if (memoryPoolFactory != null) {
             return tableWrite.withMemoryPoolFactory(memoryPoolFactory);
         } else {
-            return tableWrite.withMemoryPool(
-                    memoryPool != null
-                            ? memoryPool
-                            : new HeapMemorySegmentPool(
-                                    table.coreOptions().writeBufferSize(),
-                                    table.coreOptions().pageSize()));
+            return (TableWriteImpl<?>)
+                    tableWrite.withMemoryPool(
+                            memoryPool != null
+                                    ? memoryPool
+                                    : new HeapMemorySegmentPool(
+                                            table.coreOptions().writeBufferSize(),
+                                            table.coreOptions().pageSize()));
         }
     }
 
     public void withCompactExecutor(ExecutorService compactExecutor) {
         write.withCompactExecutor(compactExecutor);
     }
 
+    @Override
+    public void setWriteRestore(WriteRestore writeRestore) {
+        this.write.withWriteRestore(writeRestore);
+    }
+
     @Override
     public SinkRecord write(InternalRow internalRow) throws Exception {
         return write.writeAndReturn(internalRow);

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java

@@ -184,7 +184,7 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
                         bucket = 0;
                         break;
                     }
-                case CROSS_PARTITION:
+                case KEY_DYNAMIC:
                 default:
                     {
                         throw new RuntimeException("Unsupported bucket mode: " + tuple4.f0);