diff --git a/tez-dag/src/main/java/org/apache/tez/dag/history/utils/DAGUtils.java b/tez-dag/src/main/java/org/apache/tez/dag/history/utils/DAGUtils.java
index 6cadd2e96d..e028ed60da 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/history/utils/DAGUtils.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/history/utils/DAGUtils.java
@@ -30,6 +30,7 @@
 import java.util.TreeMap;
 import java.util.zip.Inflater;
 
+import org.apache.hadoop.conf.ConfigRedactor;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.tez.common.ATSConstants;
 import org.apache.tez.common.Preconditions;
@@ -505,11 +506,15 @@ public static Map<String,Object> convertEdgeManagerPluginDescriptor(
 
   public static Map<String, String> convertConfigurationToATSMap(Configuration conf) {
     // Copy configuration to avoid CME since iterator is not thread safe until HADOOP-13500
-    Iterator<Entry<String, String>> iter = new Configuration(conf).iterator();
+    Configuration snapshot = new Configuration(conf);
+    // Mask values of keys matching hadoop.security.sensitive-config-keys
+    // before publishing the configuration to the history service.
+    ConfigRedactor redactor = new ConfigRedactor(snapshot);
+    Iterator<Entry<String, String>> iter = snapshot.iterator();
     Map<String, String> atsConf = new TreeMap<String, String>();
     while (iter.hasNext()) {
       Entry<String, String> entry = iter.next();
-      atsConf.put(entry.getKey(), entry.getValue());
+      atsConf.put(entry.getKey(), redactor.redact(entry.getKey(), entry.getValue()));
     }
     return atsConf;
   }
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/history/utils/TestDAGUtils.java b/tez-dag/src/test/java/org/apache/tez/dag/history/utils/TestDAGUtils.java
index 2bf3746bbb..70356a1695 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/history/utils/TestDAGUtils.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/history/utils/TestDAGUtils.java
@@ -31,6 +31,9 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.security.ssl.FileBasedKeyStoresFactory;
+import org.apache.hadoop.security.ssl.SSLFactory;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.tez.client.CallerContext;
@@ -47,8 +50,10 @@
 import org.apache.tez.dag.api.OutputCommitterDescriptor;
 import org.apache.tez.dag.api.OutputDescriptor;
 import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
 import org.apache.tez.dag.api.Vertex;
 import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
+import org.apache.tez.dag.app.dag.impl.DAGSchedulerNaturalOrder;
 import org.apache.tez.dag.records.TezDAGID;
 import org.apache.tez.dag.records.TezVertexID;
 import org.apache.tez.runtime.api.OutputCommitter;
@@ -222,4 +227,52 @@ public void testConvertDAGPlanToATSMap() throws IOException, JSONException {
     }
   }
 
+  @Test
+  public void testConvertConfigurationToATSMapRedactsSecrets() {
+    // Defaults must load: ConfigRedactor's fallback pattern is narrower than
+    // core-default.xml and misses fs.s3a.session.token.
+    Configuration conf = new Configuration();
+    String serverKeystorePassword = FileBasedKeyStoresFactory.resolvePropertyName(
+        SSLFactory.Mode.SERVER, FileBasedKeyStoresFactory.SSL_KEYSTORE_PASSWORD_TPL_KEY);
+    String clientTruststorePassword = FileBasedKeyStoresFactory.resolvePropertyName(
+        SSLFactory.Mode.CLIENT, FileBasedKeyStoresFactory.SSL_TRUSTSTORE_PASSWORD_TPL_KEY);
+    conf.set(TezConfiguration.TEZ_AM_DAG_SCHEDULER_CLASS, DAGSchedulerNaturalOrder.class.getName());
+    conf.set(TezConfiguration.TEZ_QUEUE_NAME, "normal-queue");
+    // S3A key constants live in hadoop-aws, which is not a tez-dag dependency.
+    conf.set("fs.s3a.secret.key", "fake-s3a-secret-key");
+    conf.set("fs.s3a.session.token", "fake-s3a-session-token");
+    conf.set(serverKeystorePassword, "fake-keystore-password");
+    conf.set(clientTruststorePassword, "fake-truststore-password");
+    // An access key id is not a secret; the default pattern omits it.
+    conf.set("fs.s3a.access.key", "fake-s3a-access-key-id");
+
+    Map<String, String> ats = DAGUtils.convertConfigurationToATSMap(conf);
+
+    assertEquals(DAGSchedulerNaturalOrder.class.getName(),
+        ats.get(TezConfiguration.TEZ_AM_DAG_SCHEDULER_CLASS));
+    assertEquals("normal-queue", ats.get(TezConfiguration.TEZ_QUEUE_NAME));
+    assertEquals("fake-s3a-access-key-id", ats.get("fs.s3a.access.key"));
+    for (String key : new String[] {"fs.s3a.secret.key", "fs.s3a.session.token",
+        serverKeystorePassword, clientTruststorePassword}) {
+      assertFalse(ats.get(key).contains("fake"),
+          key + " must be redacted, got: " + ats.get(key));
+    }
+  }
+
+  @Test
+  public void testConvertConfigurationToATSMapHonoursConfiguredPattern() {
+    // A configured pattern replaces the defaults and must be honoured.
+    Configuration conf = new Configuration();
+    conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_SENSITIVE_CONFIG_KEYS,
+        "my\\.site\\.internal\\.id");
+    conf.set("my.site.internal.id", "fake-internal-id");
+    conf.set(TezConfiguration.TEZ_QUEUE_NAME, "normal-queue");
+
+    Map<String, String> ats = DAGUtils.convertConfigurationToATSMap(conf);
+
+    assertFalse(ats.get("my.site.internal.id").contains("fake"),
+        "configured sensitive key must be redacted, got: " + ats.get("my.site.internal.id"));
+    assertEquals("normal-queue", ats.get(TezConfiguration.TEZ_QUEUE_NAME));
+  }
+
 }