@Test
public void testPropertyRedactionForJHS() throws Exception {

  final Configuration conf = new Configuration();
  String sensitivePropertyName = "aws.fake.credentials.name";
  String sensitivePropertyValue = "aws.fake.credentials.val";
  conf.set(sensitivePropertyName, sensitivePropertyValue);
  conf.set(MRJobConfig.MR_JOB_REDACTED_PROPERTIES,
      sensitivePropertyName);
  conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY,
      dfsCluster.getURI().toString());
  final TestParams params = new TestParams();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, params.dfsWorkDir);
  final JHEvenHandlerForTest jheh =
      new JHEvenHandlerForTest(params.mockAppContext, 0, false);
  try {
    jheh.init(conf);
    jheh.start();
    handleEvent(jheh, new JobHistoryEvent(params.jobId,
        new AMStartedEvent(params.appAttemptId, 200, params.containerId,
            "nmhost", 3000, 4000, -1)));
    handleEvent(jheh, new JobHistoryEvent(params.jobId,
        new JobUnsuccessfulCompletionEvent(TypeConverter.fromYarn(
            params.jobId), 0, 0, 0, 0, 0, 0, 0,
            JobStateInternal.FAILED.toString())));
    // verify the value of the sensitive property in job.xml is restored.
    assertThat(conf.get(sensitivePropertyName))
        .isEqualTo(sensitivePropertyValue)
        .withFailMessage(sensitivePropertyName + " is modified.");
    // load the job_conf.xml in JHS directory and verify property redaction.
    Path jhsJobConfFile = getJobConfInIntermediateDoneDir(conf, params.jobId);
    Assert.assertTrue("The job_conf.xml file is not in the JHS directory",
        FileContext.getFileContext(conf).util().exists(jhsJobConfFile));
    Configuration jhsJobConf = new Configuration();
    try (InputStream input = FileSystem.get(conf).open(jhsJobConfFile)) {
      jhsJobConf.addResource(input);
      Assert.assertEquals(
          sensitivePropertyName + " is not redacted in HDFS.",
          MRJobConfUtil.REDACTION_REPLACEMENT_VAL,
          jhsJobConf.get(sensitivePropertyName));
    }
  } finally {
    jheh.stop();
    purgeHdfsHistoryIntermediateDoneDirectory(conf);
  }
}