[cdc] Combine mode should throw better exception to dynamic bucket ta…

…ble (#3291)
apache · May 6, 2024 · 3112c97 · 3112c97
1 parent 93666c2
commit 3112c97
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 1 deletion.
diff --git a/...cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java b/...cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.ChannelComputer;
 
@@ -77,6 +78,14 @@ private ChannelComputer<CdcRecord> computeChannelComputer(CdcMultiplexRecord rec
                         LOG.error("Failed to get table " + id.getFullName());
                         return null;
                     }
+
+                    if (table.bucketMode() != BucketMode.FIXED) {
+                        throw new UnsupportedOperationException(
+                                String.format(
+                                        "Combine mode Sink only supports FIXED bucket mode, but %s is %s",
+                                        table.name(), table.bucketMode()));
+                    }
+
                     CdcRecordChannelComputer channelComputer =
                             new CdcRecordChannelComputer(table.schema());
                     channelComputer.setup(numChannels);

diff --git a/...-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java b/...-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
@@ -190,7 +190,9 @@ private FileStoreTable getTable(Identifier tableId) throws InterruptedException
 
         if (table.bucketMode() != BucketMode.FIXED) {
             throw new UnsupportedOperationException(
-                    "Unified Sink only supports FIXED bucket mode, but is " + table.bucketMode());
+                    String.format(
+                            "Combine mode Sink only supports FIXED bucket mode, but %s is %s",
+                            table.name(), table.bucketMode()));
         }
         return table;
     }