From 9c4e3b1cf0e153f2af7785f8eeb71633e829acc2 Mon Sep 17 00:00:00 2001 From: Oleksandr Nitavskyi Date: Wed, 16 Sep 2026 13:39:46 +0200 Subject: [PATCH] [flink][cdc][iceberg] Include table identifier in schema-change and CDC-commit exception messages UpdatedDataFieldsProcessFunctionBase.applySchemaChange and several other exceptions in the multi-table CDC sink and Iceberg-REST metadata-commit paths were thrown without the table identifier, even though it (or a FileStoreTable/Identifier reference) was already in scope. In a multi-table CDC sink processing many tables through shared operators, this made it impossible to tell which table caused a crash from the exception message alone, significantly slowing diagnosis during a live production incident. This appends the table identifier to each affected message. All changes are purely additive (same exception type, same trigger conditions, message text only) with no control-flow or method-signature changes. --- .../paimon/iceberg/IcebergCommitCallback.java | 18 ++++++++++++++---- .../cdc/CdcMultiplexRecordChannelComputer.java | 2 +- .../cdc/CdcRecordStoreMultiWriteOperator.java | 6 ++++-- .../UpdatedDataFieldsProcessFunctionBase.java | 4 +++- .../paimon/flink/sink/StoreCommitter.java | 3 ++- .../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 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 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 element) throws Exce } else { throw new RuntimeException( "Unable to process element. Possibly a corrupt record: " - + (logCorruptRecord ? record : "")); + + (logCorruptRecord ? record : "") + + ", 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)); } }