{"record":{"id":"ebf20ad2330871a2","repo":"apache/iceberg","slug":"for-table-with-at-exception-proces","errorCode":null,"errorMessage":"[For table {} with {}[{}] at {}]: Exception processing {}","messagePattern":"\\[For table (.+?) with (.+?)\\[(.+?)\\] at (.+?)\\]: Exception processing (.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java","lineNumber":118,"sourceCode":"\n  @Override\n  public void processElement(StreamRecord<DataFileRewriteRunner.ExecutedGroup> streamRecord) {\n    DataFileRewriteRunner.ExecutedGroup executedGroup = streamRecord.getValue();\n    try {\n      if (commitService == null) {\n        // Refresh the table to get the latest snapshot for the committer\n        table.refresh();\n\n        FlinkRewriteDataFilesCommitManager commitManager =\n            new FlinkRewriteDataFilesCommitManager(\n                table, executedGroup.snapshotId(), streamRecord.getTimestamp(), branch);\n        this.commitService = commitManager.service(executedGroup.groupsPerCommit());\n        commitService.start();\n      }\n\n      commitService.offer(executedGroup.group());\n    } catch (Exception e) {\n      LOG.warn(\n          DataFileRewritePlanner.MESSAGE_PREFIX + \"Exception processing {}\",\n          tableName,\n          taskName,\n          taskIndex,\n          streamRecord.getTimestamp(),\n          executedGroup,\n          e);\n      output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));\n      errorCounter.inc();\n    }\n  }\n\n  @Override\n  public void processWatermark(Watermark mark) throws Exception {\n    try {\n      if (commitService != null) {\n        commitService.close();\n      }","sourceCodeStart":100,"sourceCodeEnd":136,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java#L100-L136","documentation":"DataFileRewriteCommitter.processElement() feeds executed rewrite groups into the RewriteDataFilesCommitManager's CommitService. Any exception while refreshing the table, creating/starting the commit service, or offering the group is logged with the maintenance MESSAGE_PREFIX and routed to the TaskResultAggregator error stream instead of failing the operator. The message names the offending ExecutedGroup so the failed rewrite batch is identifiable.","triggerScenarios":"table.refresh() fails (catalog/network errors loading new metadata); constructing FlinkRewriteDataFilesCommitManager or starting the CommitService fails; commitService.offer(group) throws because the rewrite output files are missing or unreadable, or the group's starting snapshotId no longer exists.","commonSituations":"Rewritten data files deleted by aggressive retention or lost because the writer task's output dir was cleaned; catalog (Hive/REST/Hadoop) temporarily unreachable during refresh; concurrent expireSnapshots removing the starting snapshot; wrong branch configured for the table.","solutions":["Inspect the chained cause in the log to find the root failure (missing files vs catalog vs metadata)","Verify the rewritten data files still exist on storage and retention policies are not deleting them mid-run","Re-run the maintenance job — the error stream does not fail the job, so a subsequent trigger retries planning and rewriting","Check that the configured branch still exists and the starting snapshot was not expired concurrently","Validate catalog connectivity/credentials from the Flink TaskManager"],"exampleFix":"// before: concurrent expireSnapshots removes snapshots the rewriter depends on\nTableMaintenance.builder().add(DataFileRewriteTask.create(maintenanceConfig))\n    .add(ExpireSnapshotsTask.create(expireConfig)) // aggressive retention\n    .schedule(...);\n\n// after: keep snapshot retention window larger than the rewrite execution window\nExpireSnapshots expire = ExpireSnapshots.builder()...\n    .retainLast(1).olderThan(System.currentTimeMillis() - MIN_SNAPSHOT_RETENTION_MS)\n    .apply();","handlingStrategy":"try-catch","validationCode":"// before the maintenance run, verify the table and branch are reachable\nTableLoader tableLoader = TableLoader.fromCatalog(...);\ntableLoader.open();\nTable table = tableLoader.loadTable(); // throws early if catalog/table unavailable\nif (table.snapshot(branch) == null && !table.currentSnapshot().isNullSnapshot()) {\n  LOG.warn(\"Branch {} missing; rewrite committer may fail\", branch);\n}","typeGuard":null,"tryCatchPattern":"// the operator already routes errors to the side-output; monitor and react\nDataStream<Exception> errors = result.getSideOutput(TaskResultAggregator.ERROR_STREAM);\nerrors.process((ProcessFunction<Exception, Void>) (e, ctx) -> {\n  alertOps(\"Rewrite committer failure: \" + e.getMessage());\n  return null;\n});","preventionTips":["Ensure rewritten output files are not deleted by retention/cleanup jobs before commit","Expire snapshots with a retention window longer than the maintenance execution window","Check catalog connectivity/credentials from TaskManagers before scheduling maintenance","Monitor the per-task error-counter metric to detect recurring commit failures"],"tags":["flink","maintenance","commit","iceberg"],"backgroundTag":"upstream-api-error","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}