Can't we store the key on KMS or encrypted HDFS so the second AM can recover it using delegation tokens?

Arun Suresh, Vinod Kumar Vavilapalli - thoughts?

The patch skips recovery if spill encryption is enabled and rerun all mappers and reducers. Have not seen any safe place to store the spill encryption key in MRAppMaster.

-1 overall

This message was automatically generated.

Uploading a new patch to address the unit test failure and check style warning

+1 overall

This message was automatically generated.

We should have two JIRAs for this - (1) Avoid recovering an AM if encrypted spill is enabled, and (2) Support recovering an AM even when encrypted spill is enabled. I am fine with using this JIRA for either, but we should file the other one too.

Comments on the patch:

1. The comment for the variable can be simplified to be clear. For instance, "When intermediate data is encrypted, recovering the job requires access to the key. Until the encryption key is persisted, we should avoid recovery attempts."
2. Can the boolean condition be simplified to numReduceTasks <= 0 || shuffleKeyValidForRecovery || !spillEncrypted?
3. I assume the new method recovered() is for tests only. Should we annotate it @VisibleForTesting?

Karthik Kambatla Thanks for your reviews! I have left this jira to do 1), so I will create another jira that does 2) after this is done. But as I said in my previous comment, MR does not seem to have any safe store to persist the encryption key across job attempts (Not sure how much efforts it is to introduce one).

Can the boolean condition be simplified to numReduceTasks <= 0 || shuffleKeyValidForRecovery || !spillEncrypted?

Not quite sure about when to make something like this incline or a tiny method. I was thinking of using the method name as comment sort of. Any guideline on this?

MR does not seem to have any safe store to persist the encryption key across job attempts

One could store the encrypted key in KMS. Once stored, we could do one of the following:

1. Tasks have a way to fetch this key directly
2. Leave the tasks as is, but augment the AM to recover this key as part of recovery.

Not quite sure about when to make something like this incline or a tiny method.

It is okay to pull this into a separate method. That said, I feel this method could do with improved logging. For instance, when we don't recover, the method dumps a bunch of information that is hard for end-users to parse. Instead, I would like something like this:

    boolean attemptRecovery = true;
    boolean recoveryEnabled = getConfig().getBoolean(
        MRJobConfig.MR_AM_JOB_RECOVERY_ENABLE,
        MRJobConfig.MR_AM_JOB_RECOVERY_ENABLE_DEFAULT);
    if (!recoveryEnabled) {
      LOG.info("Not attempting to recover. Recovery disabled. To enable " +
          "recovery, set " + MRJobConfig.MR_AM_JOB_RECOVERY_ENABLE);
      attemptRecovery = false;
    }

    boolean recoverySupportedByCommitter = isRecoverySupported();
    if (!recoverySupportedByCommitter) {
      LOG.info("Not attempting to recover. Recovery is not supported by " +
          "committer. Use an OutputCommitter that allows recovery.");
      attemptRecovery = false;
    }

    int numReduceTasks = getConfig().getInt(MRJobConfig.NUM_REDUCES, 0);
    boolean shuffleKeyValidForRecovery =
        TokenCache.getShuffleSecretKey(jobCredentials) != null;
    if (numReduceTasks > 0 && !shuffleKeyValidForRecovery) {
      LOG.info("Not attempting to recover. Shuffle key is not valid for " +
          "recovery.");
      attemptRecovery = false;
    }

    boolean recoverySucceeded = true;
    if (attemptRecovery) {
      LOG.info("Attempting to recover.");
      try {
        parsePreviousJobHistory();
      } catch (IOException e) {
        LOG.warn("Unable to parse prior job history, aborting recovery", e);
        recoverySucceeded = false;
      }
    }
    
    if (!attemptRecovery || !recoverySucceeded) {
      // Get the amInfos anyways whether recovery is enabled or not
      am

Alternatively, processRecovery could return true if recovery succeeded or if it is first attempt, and false otherwise.

Thanks for your comments, Karthik! Uploading a new patch to address your suggestions, which does make the logging much more user-friendly.

One could store the encrypted key in KMS. Once stored, we could do one of the following: 1) Tasks have a way to fetch this key directly 2) Leave the tasks as is, but augment the AM to recover this key as part of recovery.

I think option 2 is advantageous in that the change needed is minimal and only one query for each job attempt to the safe store is needed instead of one for each task.

Will create a follow up jira once this is done.

-1 overall

This message was automatically generated.

+1 overall

This message was automatically generated.

Patch looks pretty good. Really like the actionable log messages.

Couple of nits:

1. In the following log message, s/recovery/recover:

       if (reducerCount > 0 && spillEncrypted) {
         LOG.info("Not attempting to recovery. Intermediate spill encryption" +
             " is enabled.");
   

2. Consider renaming attemptRecoveryIfNotFirstAttempt to shouldAttemptRecoveryIfNotFirstAttempt. That does become quite long, which makes me think: how about shouldAttemptRecovery? This method could also check for the first attempt case and return false. And, the if condition towards the end of processRecovery could include a check for first attempt. What do you think?

Thanks for the reviews, Karthik! Uploading a new patch to address your comments.

+1, pending Jenkins.

+1 overall

This message was automatically generated.

(1) Avoid recovering an AM if encrypted spill is enabled

Encrypted spill w.r.t recovery is not the same as a committer not supporting recovery. Any reason we cannot just re-run the job from scratch if all reducers have not completed ( or re-run all maps and incomplete reducers )?

Ideally speaking, you could just re-run most of the job tasks again if needed to support proper fault tolerance even in scenarios where the key cannot be stored securely. In this scenario, the new AM can generate a new key. I would agree that this might not be a performant solution but it atleast solves the problem of not having the user to re-submit the job. If performance is an issue, users can turn off recovery when encryption is enabled for scenarios where the key cannot be stored securely.

Hi Hitesh Shah Thanks very much for your comments! Sorry for the confusing phase in there. I guess a better way to put it is probably something like
Avoid recovering progress made in previous attempts if encrypted spill is enabled.

If recovery and spill encryption are both enabled, the job attempt does not fail. The job attempt will simply start from scratch because no history file is parsed. If the naming in the patch is too misleading, I can create a new follow up jira to rename the shouldAttemptRecovery. How about shouldAttemptToRecoveryProgress?

Hitesh Shah - are you comfortable with checking this in?

Havent looked at the patch in detail but Haibo Chen's clarifying comments make sense. Jira title could be modified accordingly. +0 from my side.

Thanks Hitesh. Checking this in with the updated title.

Just committed to trunk and branch-2. Thanks Haibo Chen for cleaning this up, and Hitesh Shah for chiming in on the appropriateness of the title/commit message.

SUCCESS: Integrated in Jenkins build Hadoop-trunk-Commit #10532 (See https://builds.apache.org/job/Hadoop-trunk-Commit/10532/)
MAPREDUCE-6638. Do not attempt to recover progress from previous job (kasha: rev de7a0a92ca1983b35ca4beb7ab712fd700a9e6e0)

• (edit) hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRecovery.java
• (edit) hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java