Skip to content

[fix](fe) Mask Kafka routine load sensitive properties#64786

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
liaoxin01 merged 8 commits into from
Jul 2, 2026
Merged

[fix](fe) Mask Kafka routine load sensitive properties #64786

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
2fb017f
[fix](fe) Mask Kafka routine load sensitive properties
0AyanamiRei Jun 24, 2026
ac1f26b
[test](fe) Cover Kafka routine load secret masks
0AyanamiRei Jun 25, 2026
42ff263
Merge branch 'master' into fix-kafka-routine-load-mask-secrets
0AyanamiRei Jun 25, 2026
1ead4ed
[fix](fe) Mask Kafka OAuth routine load secrets
0AyanamiRei Jun 25, 2026
4bd30cb
[fix](fe) Mask bare Kafka routine load secrets
0AyanamiRei Jun 29, 2026
ecf8e30
[fix](fe) Fix Kafka routine load import order
0AyanamiRei Jun 29, 2026
043faf9
[fix](fe) Mask Kafka AWS access key in routine load display
0AyanamiRei Jun 30, 2026
27a837f
[fix](fe) Mask Kinesis AWS access key in routine load display
0AyanamiRei Jun 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 29 additions & 4 deletions fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -33,6 +33,7 @@
import org.apache.doris.common.LoadException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DatasourcePrintableMap;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.common.util.LogBuilder;
Expand DownExpand Up@@ -82,6 +83,7 @@
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;
Expand All@@ -102,6 +104,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;
Expand DownExpand Up@@ -723,7 +726,31 @@ public String dataSourcePropertiesJsonToString() {
@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) -> {
Comment thread
0AyanamiRei marked this conversation as resolved.
Comment thread
sollhui marked this conversation as resolved.
String lowerKey = key.toLowerCase(Locale.ROOT);
boolean sensitive = KafkaConfiguration.SASL_JAAS_CONFIG.equalsIgnoreCase(key)
Comment thread
0AyanamiRei marked this conversation as resolved.
|| KafkaConfiguration.AWS_ACCESS_KEY.equalsIgnoreCase(key)
|| DatasourcePrintableMap.SENSITIVE_KEY.contains(key)
Comment thread
0AyanamiRei marked this conversation as resolved.
|| 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
Expand All@@ -736,9 +763,7 @@ public Map<String, String> getDataSourceProperties() {

@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
Expand Down
30 changes: 14 additions & 16 deletions ...e-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,7 @@
*/
public class KinesisRoutineLoadJob extends RoutineLoadJob {
private static final Logger LOG = LogManager.getLogger(KinesisRoutineLoadJob.class);
private static final String SENSITIVE_PROPERTY_MASK = "******";

public static final String KINESIS_FILE_CATALOG = "kinesis";

Expand DownExpand Up@@ -640,15 +641,7 @@ public String dataSourcePropertiesJsonToString() {
@Override
public String customPropertiesJsonToString() {
Gson gson = new GsonBuilder().disableHtmlEscaping().create();
// Mask sensitive information
Map<String, String> maskedProperties = new HashMap<>(customProperties);
if (maskedProperties.containsKey("aws.secret_key")) {
maskedProperties.put("aws.secret_key", "******");
}
if (maskedProperties.containsKey("aws.session_key")) {
maskedProperties.put("aws.session_key", "******");
}
return gson.toJson(maskedProperties);
return gson.toJson(getMaskedCustomProperties(""));
}

@Override
Expand All@@ -664,16 +657,21 @@ public Map<String, String> getDataSourceProperties() {

@Override
public Map<String, String> getCustomProperties() {
Map<String, String> ret = new HashMap<>();
customProperties.forEach((k, v) -> {
// Mask sensitive values
if (k.equals("aws.secret_key") || k.equals("aws.session_key")) {
ret.put("property." + k, "******");
return getMaskedCustomProperties("property.");
}

private Map<String, String> getMaskedCustomProperties(String keyPrefix) {
Map<String, String> maskedProperties = new HashMap<>();
customProperties.forEach((key, value) -> {
if (KinesisConfiguration.KINESIS_ACCESS_KEY.getName().equalsIgnoreCase(key)
|| KinesisConfiguration.KINESIS_SECRET_KEY.getName().equalsIgnoreCase(key)
|| KinesisConfiguration.KINESIS_SESSION_TOKEN.getName().equalsIgnoreCase(key)) {
maskedProperties.put(keyPrefix + key, SENSITIVE_PROPERTY_MASK);
} else {
ret.put("property." + k, v);
maskedProperties.put(keyPrefix + key, value);
}
});
return ret;
return maskedProperties;
}

@Override
Expand Down
81 changes: 81 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -261,6 +261,87 @@ public void testUpdateProgressWarnsWhenReadCommittedTaskHasZeroRowsAndLag() thro
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() throws UserException {
RoutineLoadManager routineLoadManager = Mockito.mock(RoutineLoadManager.class);
Expand Down
28 changes: 28 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/load/routineload/KinesisRoutineLoadJobTest.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -311,6 +311,34 @@ public void testFullyConsumedClosedParentShouldNotReappearOnRefresh() throws Exc
Assert.assertFalse((Boolean) Deencapsulation.invoke(routineLoadJob, "isKinesisShardsChanged"));
}

@Test
public void testDisplayCustomPropertiesMasksKinesisSecrets() {
KinesisRoutineLoadJob routineLoadJob =
new KinesisRoutineLoadJob(1L, "kinesis_routine_load_job", 1L,
1L, "ap-southeast-1", "stream-1", UserIdentity.ADMIN);
Map<String, String> customProperties = Maps.newHashMap();
customProperties.put(KinesisConfiguration.KINESIS_ACCESS_KEY.getName(), "aws_access_key");
customProperties.put(KinesisConfiguration.KINESIS_SECRET_KEY.getName(), "aws_secret");
customProperties.put(KinesisConfiguration.KINESIS_SESSION_TOKEN.getName(), "aws_session_secret");
customProperties.put("aws.role_arn", "role_arn_value");
Deencapsulation.setField(routineLoadJob, "customProperties", customProperties);

String customPropertiesJson = routineLoadJob.customPropertiesJsonToString();
Map<String, String> showCreateCustomProperties = routineLoadJob.getCustomProperties();

Assert.assertFalse(customPropertiesJson.contains("aws_access_key"));
Assert.assertFalse(customPropertiesJson.contains("aws_secret"));
Assert.assertFalse(customPropertiesJson.contains("aws_session_secret"));
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("\"aws.role_arn\":\"role_arn_value\""));
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("role_arn_value", showCreateCustomProperties.get("property.aws.role_arn"));
}

private Set<String> collectAssignedShards(KinesisRoutineLoadJob routineLoadJob) {
List<RoutineLoadTaskInfo> routineLoadTaskInfoList =
Deencapsulation.getField(routineLoadJob, "routineLoadTaskInfoList");
Expand Down
2 changes: 1 addition & 1 deletion regression-test/suites/load_p0/kinesis_routine_load/test_kinesis_show_routine_load.groovy
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -224,7 +224,7 @@ suite("test_kinesis_show_routine_load") {

def customProperties = parseJson(showRow[13].toString())
assertEquals("******", customProperties["aws.secret_key"].toString())
assertTrue(customProperties["aws.access_key"].toString().length() > 0)
assertEquals("******", customProperties["aws.access_key"].toString())

// Verify additional SHOW ROUTINE LOAD paths.
String dbName = context.config.getDbNameByFile(context.file)
Expand Down
41 changes: 41 additions & 0 deletions regression-test/suites/load_p0/routine_load/test_show_routine_load.groovy
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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 """
Expand Down
Loading