This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit cdea2a1a9c41f5cd9dbb4f14c6967d9dd56ca878 Author: Oleksandr Nitavskyi <[email protected]> AuthorDate: Thu Sep 17 07:12:05 2026 +0200 [flink][cdc][iceberg] Include table identifier in schema-change and CDC-commit exception messages (#9892) --- .../apache/paimon/iceberg/IcebergCommitCallback.java | 18 ++++++++++++++---- .../sink/cdc/CdcMultiplexRecordChannelComputer.java | 2 +- .../sink/cdc/CdcRecordStoreMultiWriteOperator.java | 6 ++++-- .../sink/cdc/UpdatedDataFieldsProcessFunctionBase.java | 4 +++- .../org/apache/paimon/flink/sink/StoreCommitter.java | 3 ++- .../paimon/iceberg/IcebergRestMetadataCommitter.java | 5 +++-- 6 files changed, 27 insertions(+), 11 deletions(-) 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 5b0a78b828..ab73175396 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 @@ public class IcebergCommitCallback implements CommitCallback, TagCallback { } 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 @@ public class IcebergCommitCallback implements CommitCallback, TagCallback { } 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 class IcebergCommitCallback implements CommitCallback, TagCallback { 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 class IcebergCommitCallback implements CommitCallback, TagCallback { 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 053108f1cf..07dbb3f9f0 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 @@ public class CdcMultiplexRecordChannelComputer implements ChannelComputer<CdcMul } 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 20dd9847c4..20560ea878 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 class CdcRecordStoreMultiWriteOperator } 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 877fc8d20b..1841d57aea 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 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends Process "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 4c353517c3..8bc79914d4 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 class StoreCommitter implements Committer<Committable, ManifestCommittabl 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 1115624788..aafd741439 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 @@ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter { || 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)); } }
