This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8f43d09fc3 [flink][cdc][iceberg] Include table identifier in
schema-change and CDC-commit exception messages (#9892)
8f43d09fc3 is described below
commit 8f43d09fc3ccba076209d39b1365369b8d6d9e1d
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));
}
}
