This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new ef5567547ed branch-4.1: [fix](fe) Mask Kafka routine load sensitive
properties #64786 (#65629)
ef5567547ed is described below
commit ef5567547ede5d9ac2cffe100c2e4f10556efb9f
Author: Refrain <[email protected]>
AuthorDate: Thu Jul 16 11:11:11 2026 +0800
branch-4.1: [fix](fe) Mask Kafka routine load sensitive properties #64786
(#65629)
### What problem does this PR solve?
Issue Number: None
Related PR: #64786
Problem Summary: Backport Kafka routine-load sensitive-property masking
from #64786 to `branch-4.1`. The display paths for `SHOW ROUTINE LOAD`,
`SHOW CREATE ROUTINE LOAD`, and `information_schema.routine_load_jobs`
mask credentials while the internal Kafka client properties remain
unchanged.
Source merge commit: `c4ddfbaa66a6039a0383fd3fa1fb8c549bfa4381`.
### Conflict resolution
- `branch-4.1` does not contain Kinesis routine load. Kept
`KinesisRoutineLoadJob`, its unit test, and its regression suite absent
instead of resurrecting a component unavailable on the target branch.
- Preserved the branch's JMockit-based `KafkaRoutineLoadJobTest` and
inserted only the source PR's sensitive-property masking test.
- Replaced master's `DatasourcePrintableMap.SENSITIVE_KEY` with the
equivalent `PrintableMap.SENSITIVE_KEY` registry available on
branch-4.1.
- Retained the branch's existing Kafka configuration constants for SASL
JAAS and AWS access-key matching.
- The Kafka regression test merged against the target suite without
additional semantic changes.
### Release note
Mask sensitive Kafka routine load custom properties in metadata display
output.
### Check List (For Author)
- Test: No need to test (one-time Agent conflict-resolution workflow;
build and test execution were intentionally excluded)
- `git diff HEAD^ HEAD --check`
- Verified no conflict markers or unmerged index entries remain
- Behavior changed: Yes (Kafka routine load credentials are masked in
external metadata displays)
- Does this need documentation: No
---
.../load/routineload/KafkaRoutineLoadJob.java | 33 +++++++--
.../load/routineload/KafkaRoutineLoadJobTest.java | 81 ++++++++++++++++++++++
.../routine_load/test_show_routine_load.groovy | 41 +++++++++++
3 files changed, 151 insertions(+), 4 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
index 36e09f01e32..cb563e725db 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
@@ -34,6 +34,7 @@ import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.common.util.LogBuilder;
import org.apache.doris.common.util.LogKey;
+import org.apache.doris.common.util.PrintableMap;
import org.apache.doris.common.util.SmallFileMgr;
import org.apache.doris.common.util.SmallFileMgr.SmallFile;
import org.apache.doris.common.util.TimeUtils;
@@ -75,6 +76,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.TimeZone;
import java.util.UUID;
@@ -95,6 +97,7 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
private static final String READ_COMMITTED_ZERO_ROWS_WITH_LAG_MESSAGE =
"Kafka routine load consumed 0 rows "
+ "while lag is still positive under
isolation.level=read_committed. If the upstream producer uses "
+ "Kafka transactions, some records may be in uncommitted
transactions and are not visible yet.";
+ private static final String SENSITIVE_PROPERTY_MASK = "******";
@SerializedName("bl")
private String brokerList;
@@ -710,7 +713,31 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
@Override
public String customPropertiesJsonToString() {
Gson gson = new GsonBuilder().disableHtmlEscaping().create();
- return gson.toJson(customProperties);
+ return gson.toJson(getMaskedCustomProperties(""));
+ }
+
+ private Map<String, String> getMaskedCustomProperties(String keyPrefix) {
+ Map<String, String> maskedProperties = new HashMap<>();
+ customProperties.forEach((key, value) -> {
+ String lowerKey = key.toLowerCase(Locale.ROOT);
+ boolean sensitive =
KafkaConfiguration.SASL_JAAS_CONFIG.equalsIgnoreCase(key)
+ || KafkaConfiguration.AWS_ACCESS_KEY.equalsIgnoreCase(key)
+ || PrintableMap.SENSITIVE_KEY.contains(key)
+ || lowerKey.endsWith(".password")
+ || lowerKey.endsWith(".secret")
+ || lowerKey.endsWith(".secret_key")
+ || lowerKey.endsWith(".secret.key")
+ || lowerKey.endsWith(".session_key")
+ || lowerKey.endsWith(".session.token")
+ || lowerKey.contains(".private.key.")
+ || lowerKey.endsWith(".private.key")
+ || lowerKey.endsWith(".private_key")
+ || lowerKey.endsWith(".passphrase")
+ || "ssl.keystore.key".equals(lowerKey)
+ || "ssl.key.pem".equals(lowerKey);
+ maskedProperties.put(keyPrefix + key, sensitive ?
SENSITIVE_PROPERTY_MASK : value);
+ });
+ return maskedProperties;
}
@Override
@@ -723,9 +750,7 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
@Override
public Map<String, String> getCustomProperties() {
- Map<String, String> ret = new HashMap<>();
- customProperties.forEach((k, v) -> ret.put("property." + k, v));
- return ret;
+ return getMaskedCustomProperties("property.");
}
@Override
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
index b6944df5539..2565f26a15b 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
@@ -256,6 +256,87 @@ public class KafkaRoutineLoadJobTest {
Assert.assertTrue(otherMsg.contains("some records may be in
uncommitted transactions"));
}
+ @Test
+ public void testDisplayCustomPropertiesMasksKafkaSecrets() {
+ KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L,
"kafka_routine_load_job", 1L,
+ 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+ Map<String, String> customProperties = Maps.newHashMap();
+ customProperties.put("security.protocol", "SASL_PLAINTEXT");
+ customProperties.put("sasl.username", "doris");
+ customProperties.put("sasl.password", "plain_secret");
+ customProperties.put("sasl.jaas.config", "username=\"doris\"
password=\"jaas_secret\"");
+ customProperties.put("sasl.oauthbearer.client.secret",
"oauth_client_secret");
+
customProperties.put("sasl.oauthbearer.client.credentials.client.secret",
"oauth_alias_secret");
+ customProperties.put("sasl.oauthbearer.assertion.private.key.pem",
"oauth_private_key_pem");
+
customProperties.put("sasl.oauthbearer.assertion.private.key.passphrase",
"oauth_private_key_passphrase");
+ customProperties.put("ssl.keystore.password", "keystore_secret");
+ customProperties.put("ssl.keystore.key", "keystore_key_secret");
+ customProperties.put("ssl.key.pem", "key_pem_secret");
+ customProperties.put(KafkaConfiguration.AWS_ACCESS_KEY,
"aws_access_key");
+ customProperties.put("aws.secret_key", "aws_secret");
+ customProperties.put("aws.session_key", "aws_session_secret");
+ customProperties.put("password", "bare_password_secret");
+ customProperties.put("secret_key", "bare_secret_key");
+ customProperties.put("session_token", "bare_session_token");
+ Deencapsulation.setField(routineLoadJob, "customProperties",
customProperties);
+
+ String customPropertiesJson =
routineLoadJob.customPropertiesJsonToString();
+ Map<String, String> showCreateCustomProperties =
routineLoadJob.getCustomProperties();
+
+ Assert.assertFalse(customPropertiesJson.contains("plain_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("jaas_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_client_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_alias_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_private_key_pem"));
+
Assert.assertFalse(customPropertiesJson.contains("oauth_private_key_passphrase"));
+ Assert.assertFalse(customPropertiesJson.contains("keystore_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("keystore_key_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("key_pem_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("aws_access_key"));
+ Assert.assertFalse(customPropertiesJson.contains("aws_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("aws_session_secret"));
+
Assert.assertFalse(customPropertiesJson.contains("bare_password_secret"));
+ Assert.assertFalse(customPropertiesJson.contains("bare_secret_key"));
+
Assert.assertFalse(customPropertiesJson.contains("bare_session_token"));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.password\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.jaas.config\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.oauthbearer.client.secret\":\"******\""));
+ Assert.assertTrue(customPropertiesJson.contains(
+
"\"sasl.oauthbearer.client.credentials.client.secret\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"sasl.oauthbearer.assertion.private.key.pem\":\"******\""));
+ Assert.assertTrue(customPropertiesJson.contains(
+
"\"sasl.oauthbearer.assertion.private.key.passphrase\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"ssl.keystore.password\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"ssl.keystore.key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"ssl.key.pem\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.access_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.secret_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"aws.session_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"password\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"secret_key\":\"******\""));
+
Assert.assertTrue(customPropertiesJson.contains("\"session_token\":\"******\""));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.sasl.password"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.sasl.jaas.config"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.sasl.oauthbearer.client.secret"));
+ Assert.assertEquals("******",
+
showCreateCustomProperties.get("property.sasl.oauthbearer.client.credentials.client.secret"));
+ Assert.assertEquals("******",
+
showCreateCustomProperties.get("property.sasl.oauthbearer.assertion.private.key.pem"));
+ Assert.assertEquals("******",
+
showCreateCustomProperties.get("property.sasl.oauthbearer.assertion.private.key.passphrase"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.ssl.keystore.password"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.ssl.keystore.key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.ssl.key.pem"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.access_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.secret_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.aws.session_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.password"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.secret_key"));
+ Assert.assertEquals("******",
showCreateCustomProperties.get("property.session_token"));
+ Assert.assertEquals("doris",
showCreateCustomProperties.get("property.sasl.username"));
+ Assert.assertEquals("plain_secret",
customProperties.get("sasl.password"));
+ }
+
@Test
public void testReadCommittedZeroRowsWithLagDelaysNextTask(
@Injectable RoutineLoadManager routineLoadManager) throws
UserException {
diff --git
a/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
b/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
index f89e89a9dda..fc330bb7b1d 100644
--- a/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
+++ b/regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
@@ -204,6 +204,47 @@ suite("test_show_routine_load","p0") {
sql "stop routine load for testShow"
}
+ // test show routine load and information_schema mask sensitive custom
properties
+ try {
+ def kafkaPassword = "doris_routine_load_password_should_be_hidden"
+ def kafkaJaasSecret =
"doris_routine_load_jaas_secret_should_be_hidden"
+ sql """
+ CREATE ROUTINE LOAD testShowSensitiveProperties ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${externalEnvIp}:${kafka_port}",
+ "kafka_topic" = "${kafkaCsvTpoics[0]}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING",
+ "property.security.protocol" = "SASL_PLAINTEXT",
+ "property.sasl.mechanism" = "PLAIN",
+ "property.sasl.username" = "doris",
+ "property.sasl.password" = "${kafkaPassword}",
+ "property.sasl.jaas.config" = "password=${kafkaJaasSecret}"
+ );
+ """
+
+ def showResult = sql "show routine load for
testShowSensitiveProperties"
+ def showCustomProperties = parseJson(showResult[0][13])
+ assertEquals("******",
showCustomProperties["sasl.password"].toString())
+ assertEquals("******",
showCustomProperties["sasl.jaas.config"].toString())
+ assertFalse(showResult[0][13].toString().contains(kafkaPassword))
+ assertFalse(showResult[0][13].toString().contains(kafkaJaasSecret))
+
+ def systemTableResult = sql """
+ SELECT CUSTOM_PROPERTIES
+ FROM information_schema.routine_load_jobs
+ WHERE JOB_NAME = 'testShowSensitiveProperties'
+ """
+ def systemTableCustomProperties =
parseJson(systemTableResult[0][0])
+ assertEquals("******",
systemTableCustomProperties["sasl.password"].toString())
+ assertEquals("******",
systemTableCustomProperties["sasl.jaas.config"].toString())
+
assertFalse(systemTableResult[0][0].toString().contains(kafkaPassword))
+
assertFalse(systemTableResult[0][0].toString().contains(kafkaJaasSecret))
+ } finally {
+ sql "stop routine load for testShowSensitiveProperties"
+ }
+
// test show routine load computegroup
try {
sql """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
