Close apache#3027

Pandas886 · Mar 15, 2024 · 68a1896 · 68a1896
1 parent 35acc4c
commit 68a1896
Showing 1 changed file with 8 additions and 6 deletions.
diff --git a/...nk-common/src/main/java/org/apache/paimon/flink/sink/MultiTableCommittableSerializer.java b/...nk-common/src/main/java/org/apache/paimon/flink/sink/MultiTableCommittableSerializer.java
@@ -26,6 +26,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 
 /**
  * {@link SimpleVersionedSerializer} for {@link MultiTableCommittable}. If a type info class is
@@ -53,9 +54,9 @@ public int getVersion() {
     public byte[] serialize(MultiTableCommittable committable) throws IOException {
         // first serialize all metadata
         String database = committable.getDatabase();
-        int databaseLen = database.length();
+        int databaseLen = database.getBytes(StandardCharsets.UTF_8).length;
         String table = committable.getTable();
-        int tableLen = table.length();
+        int tableLen = table.getBytes(StandardCharsets.UTF_8).length;
 
         int multiTableMetaLen = databaseLen + tableLen + 2 * 4;
 
@@ -83,17 +84,18 @@ public MultiTableCommittable deserialize(int committableVersion, byte[] bytes)
         int databaseLen = buffer.getInt();
         byte[] databaseBytes = new byte[databaseLen];
         buffer.get(databaseBytes, 0, databaseLen);
-        String database = new String(databaseBytes);
+        String database = new String(databaseBytes,StandardCharsets.UTF_8);
+
         int tableLen = buffer.getInt();
         byte[] tableBytes = new byte[tableLen];
         buffer.get(tableBytes, 0, tableLen);
-        String table = new String(tableBytes);
-        int multiTableMetaLen = databaseLen + tableLen + 2 * 4;
+        String table = new String(tableBytes,StandardCharsets.UTF_8);
+        int multiTableMetaLen = 4 + databaseLen + 4 + tableLen;
 
         // use committable serializer (of the same version) to deserialize committable
         byte[] serializedCommittable = new byte[bytes.length - multiTableMetaLen];
 
-        buffer.get(serializedCommittable, 0, bytes.length - multiTableMetaLen);
+        buffer.get(serializedCommittable, 0, serializedCommittable.length);
         Committable committable = deserializeCommittable(committableVersion, serializedCommittable);
 
         return MultiTableCommittable.fromCommittable(