{"record":{"id":"98d77be9a59dd7b2","repo":"apache/hadoop","slug":"unable-to-recover-task-s-output-s","errorCode":null,"errorMessage":"Unable to recover task %s, output: %s","messagePattern":"Unable to recover task (.+?), output: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"hadoop-cloud-storage-project/hadoop-tos/src/main/java/org/apache/hadoop/fs/tosfs/commit/Committer.java","lineNumber":441,"sourceCode":"            } catch (FileNotFoundException e) {\n              LOG.debug(\"Listed file already deleted: {}\", f);\n            } catch (IOException e) {\n              throw new UncheckedIOException(e);\n            } finally {\n              final FileStatus pendingFile = f;\n              CommonUtils.runQuietly(() -> destFs.delete(pendingFile.getPath(), false));\n            }\n          });\n    } finally {\n      CommonUtils.runQuietly(() -> destFs.delete(taskAttemptBasePath, true));\n    }\n  }\n\n  @Override\n  public void recoverTask(TaskAttemptContext context) {\n    checkJobId(context);\n    String taskId = context.getTaskAttemptID().toString();\n    throw new UnsupportedOperationException(\n        String.format(\"Unable to recover task %s, output: %s\", taskId, outputPath));\n  }\n\n  private int commitThreads() {\n    return conf.getInt(COMMITTER_THREADS, DEFAULT_COMMITTER_THREADS);\n  }\n\n  private void cleanup(ExecutorService pool, boolean suppress) {\n    LOG.info(\"Cleanup the job by abort the multipart uploads and clean staging dir, suppress {}\",\n        suppress);\n    try {\n      Path jobOutput = getOutputPath();\n      Iterable<MultipartUpload> pending = storage.listUploads(\n          ObjectUtils.pathToKey(CommitUtils.magicJobPath(jobId, jobOutput), true));\n      Tasks.foreach(pending)\n          .executeWith(pool)\n          .suppressFailureWhenFinished()\n          .run(u -> storage.abortMultipartUpload(u.key(), u.uploadId()));","sourceCodeStart":423,"sourceCodeEnd":459,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-cloud-storage-project/hadoop-tos/src/main/java/org/apache/hadoop/fs/tosfs/commit/Committer.java#L423-L459","documentation":"The mapreduce-API TOS staging committer (org.apache.hadoop.fs.tosfs.commit.Committer) deliberately does not implement task-output recovery: recoverTask() unconditionally throws UnsupportedOperationException. The trap is that the committer inherits the default OutputCommitter.isRecoverySupported() == true, so the MR framework believes recovery works and calls the method, which then kills the job.","triggerScenarios":"The MR ApplicationMaster restarts (YARN RM restart, AM preemption, node loss) while yarn.app.mapreduce.am.job.recovery.enable is true (the default). If a task attempt had already committed output before the restart, the restarted AM invokes committer.recoverTask(context) for the old attempt and this UnsupportedOperationException is thrown.","commonSituations":"Jobs writing to tos:// URIs through the TOS FileOutputCommitter experiencing an AM restart; clusters with YARN RM recovery enabled; users porting jobs from FileOutputCommitter (which does implement recoverTask) to the TOS committer and assuming the same recovery semantics.","solutions":["Set yarn.app.mapreduce.am.job.recovery.enable=false in mapred-site.xml (or on the job configuration) so the framework never attempts task recovery with this committer","Subclass the TOS Committer, override isRecoverySupported() to return false, and register the subclass via the output-committer factory so the framework skips recovery cleanly","Re-run the job from scratch: abortJob will clean any pending uploads/staging state left behind, so a fresh run is safe","If recovery is a hard requirement, implement recoverTask in a custom committer that remaps the previous attempt's pending-set files instead of throwing"],"exampleFix":"// before: committer inherits isRecoverySupported()==true, framework calls recoverTask(), job dies\npublic class MyCommitter extends org.apache.hadoop.fs.tosfs.commit.Committer {\n  // recoverTask() inherited -> throws UnsupportedOperationException on AM restart\n}\n// after: declare recovery unsupported so the MR framework never calls recoverTask\npublic class MyCommitter extends org.apache.hadoop.fs.tosfs.commit.Committer {\n  @Override\n  public boolean isRecoverySupported() {\n    return false;\n  }\n}","handlingStrategy":"validation","validationCode":"// before submitting a job whose output goes through the TOS committer:\nConfiguration conf = job.getConfiguration();\nString amRecovery = \"yarn.app.mapreduce.am.job.recovery.enable\"; // MRJobConfig.MR_AM_JOB_RECOVERY_ENABLE\nif (conf.getBoolean(amRecovery, true)) {\n  conf.setBoolean(amRecovery, false); // TOS committer cannot recover tasks\n}","typeGuard":null,"tryCatchPattern":"try {\n  committer.recoverTask(context);\n} catch (UnsupportedOperationException e) {\n  throw new IOException(\"TOS committer does not support task recovery; \"\n      + \"disable yarn.app.mapreduce.am.job.recovery.enable and re-run the job\", e);\n}","preventionTips":["Set yarn.app.mapreduce.am.job.recovery.enable=false cluster-wide (or per job) when the output format commits through the TOS committer","Check committer.isRecoverySupported() before relying on AM-restart recovery; subclass and return false if you control the committer class","Test AM restart (kill the MR AM mid-job) in staging before assuming a tos:// job survives recovery","Monitor for upgrades of hadoop-tos: a future version may implement recoverTask"],"tags":["hadoop","tos","output-committer","mapreduce","job-recovery"],"backgroundTag":"unsupported-operation","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-22T20:17:22.307Z"}