[flink][cdc][iceberg] Include table identifier in schema-change and CDC-commit exception messages

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);
         }
     }
 

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);
                     }
                 });
     }

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);
             }
         }
     }

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);
         }
     }
 

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();

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));
         }
     }