{"record":{"id":"c82ed271a19ec178","repo":"apache/iceberg","slug":"for-table-with-at-exception-proces-c82ed2","errorCode":null,"errorMessage":"[For table {} with {}[{}] at {}]: Exception processing {}","messagePattern":"\\[For table (.+?) with (.+?)\\[(.+?)\\] at (.+?)\\]: Exception processing (.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"flink/v2.3/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.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java#L100-L136","documentation":"DataFileRewriteCommitter.processElement catches Exceptions while offering an executed rewrite group to the commit service and logs them with the maintenance message prefix (table, task, index, timestamp). The failed group is not committed; processing of subsequent records continues.","triggerScenarios":"commitService.offer(executedGroup.group()) or commitManager.service(...) setup throws — e.g. table loading failure, invalid rewrite group, or internal commit-service error when the RewriteDataFiles commit output arrives at the committer operator.","commonSituations":"Table schema/spec changed between planning and commit; concurrent commits conflicting; misconfigured maintenance job producing groups the commit manager cannot handle.","solutions":["Read the attached exception and the prefixed table/task identifiers to locate the failing rewrite task","Re-run table maintenance for the affected table once the cause is fixed","Check for concurrent competing compaction jobs or catalog commit conflicts","Validate the maintenance job configuration (groupsPerCommit, table loading options)"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// ensure the maintenance table loads before enabling rewrite job\n tableLoader.open().refresh();","typeGuard":null,"tryCatchPattern":"try {\n  commitService.offer(group);\n} catch (Exception e) {\n  LOG.warn(MESSAGE_PREFIX + \"Exception processing {}\", table, task, idx, ts, e);\n  // re-run maintenance later\n}","preventionTips":["Avoid concurrent competing compaction jobs on the same table","Validate maintenance job configuration before deploying","Monitor logs for this warning and re-run affected rewrite tasks"],"tags":["flink","maintenance","compaction","commit"],"backgroundTag":"maintenance-task-failed","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}