diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index 5b0a78b828fd..ab731753969e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
@@ -632,7 +632,11 @@ private void createMetadataWithoutBase(
         }
         if (!written && !metadataMatchesSnapshot(snapshotId, paimonSnapshot)) {
             // no twin published this snapshot's metadata; fail so the commit retries
-            throw new IllegalStateException("Failed to replace Iceberg metadata " + metadataPath);
+            throw new IllegalStateException(
+                    "Failed to replace Iceberg metadata "
+                            + metadataPath
+                            + " for table "
+                            + table.name());
         }
         // a delayed callback may still write its metadata (a newer commit extends it), but
         // only the current head may move the hint and the external catalog
@@ -1253,7 +1257,11 @@ private void createMetadataWithBase(
         }
         if (!written && !metadataMatchesSnapshot(snapshotId, snapshot)) {
             // no twin published this snapshot's metadata; fail so the commit retries
-            throw new IllegalStateException("Failed to replace Iceberg metadata " + metadataPath);
+            throw new IllegalStateException(
+                    "Failed to replace Iceberg metadata "
+                            + metadataPath
+                            + " for table "
+                            + table.name());
         }
         // a delayed callback may still write its metadata (a newer commit extends it), but
         // only the current head may move the hint and the external catalog
@@ -1739,7 +1747,8 @@ public void notifyCreation(String tagName, long snapshotId) {
                     snapshotId);
 
         } catch (IOException e) {
-            throw new UncheckedIOException("Failed to create tag " + tagName, e);
+            throw new UncheckedIOException(
+                    "Failed to create tag " + tagName + " for table " + table.name(), e);
         }
     }
 
@@ -1797,7 +1806,8 @@ public void notifyDeletion(String tagName) {
                     tagName);
 
         } catch (IOException e) {
-            throw new UncheckedIOException("Failed to create tag " + tagName, e);
+            throw new UncheckedIOException(
+                    "Failed to create tag " + tagName + " for table " + table.name(), e);
         }
     }
 
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
index 053108f1cf2f..07dbb3f9f0f0 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
@@ -109,7 +109,7 @@ private ChannelComputer<CdcRecord> computeChannelComputer(CdcMultiplexRecord rec
                     } catch (RuntimeException e) {
                         throw e;
                     } catch (Exception e) {
-                        throw new RuntimeException(e);
+                        throw new RuntimeException("Failed to compute channel for table " + id, e);
                     }
                 });
     }
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
index 20dd9847c478..20560ea87817 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
@@ -215,13 +215,15 @@ public void processElement(StreamRecord<CdcMultiplexRecord> element) throws Exce
             } else {
                 throw new RuntimeException(
                         "Unable to process element. Possibly a corrupt record: "
-                                + (logCorruptRecord ? record : "<redacted>"));
+                                + (logCorruptRecord ? record : "<redacted>")
+                                + ", table "
+                                + tableId);
             }
         } else {
             try {
                 write.write(optionalConverted.get());
             } catch (Exception e) {
-                throw new IOException(e);
+                throw new IOException("Failed to write record for table " + tableId, e);
             }
         }
     }
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
index 877fc8d20b97..1841d57aeadb 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -160,7 +160,9 @@ protected void applySchemaChange(
                     "Unsupported schema change class "
                             + schemaChange.getClass().getName()
                             + ", content "
-                            + schemaChange);
+                            + schemaChange
+                            + ", table "
+                            + identifier);
         }
     }
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
index 4c353517c3f0..8bc79914d499 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
@@ -63,7 +63,8 @@ public StoreCommitter(FileStoreTable table, TableCommit commit, Context context)
         try {
             this.commitListeners = CommitListeners.create(context, table);
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            throw new RuntimeException(
+                    "Failed to create commit listeners for table " + table.name(), e);
         }
 
         String[] tempDirs = context.tempDirs();
diff --git a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
index 1115624788dd..aafd74143915 100644
--- a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
+++ b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
@@ -512,11 +512,12 @@ private void verifyRegistered(TableMetadata newMetadata) {
                 || registered != newMetadata.currentSnapshot().snapshotId()) {
             throw new IllegalStateException(
                     String.format(
-                            "Registered catalog table is at snapshot %s instead of %s",
+                            "Registered catalog table is at snapshot %s instead of %s for table %s",
                             registered,
                             newMetadata.currentSnapshot() == null
                                     ? "null"
-                                    : newMetadata.currentSnapshot().snapshotId()));
+                                    : newMetadata.currentSnapshot().snapshotId(),
+                            icebergTableIdentifier));
         }
     }