Skip to content

Commit

Permalink
[hotfix][zeta] conf property is null, print log bug fix (#7487)
Browse files Browse the repository at this point in the history
  • Loading branch information
@hawk9821
hawk9821 authored Aug 24, 2024
1 parent 5c6a7c6 commit 2f31404
Show file tree
Hide file tree
Showing 3 changed files with 123 additions and 40 deletions.
73 changes: 36 additions & 37 deletions ...nel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -85,7 +86,7 @@ public static Config of(@NonNull Path filePath, List<String> variables) {
boolean isJson = filePath.getFileName().toString().endsWith(".json");
log.info(
"Parsed config file: \n{}",
mapToString(configDesensitization(config.root().unwrapped()), isJson));
mapToString(configDesensitization(config.root().unwrapped())));
return config;
}

Expand All @@ -107,45 +108,45 @@ public static Config of(
}
log.info(
"Parsed config file: \n{}",
mapToString(configDesensitization(config.root().unwrapped()), isJson));
mapToString(configDesensitization(config.root().unwrapped())));
return config;
}

public static Map<String, Object> configDesensitization(Map<String, Object> configMap) {
return configMap.entrySet().stream()
.collect(
Collectors.toMap(
Map.Entry::getKey,
entry -> {
String key = entry.getKey();
if (Arrays.asList(DEFAULT_SENSITIVE_KEYWORDS)
.contains(key.toLowerCase())) {
return "******";
}
Object value = entry.getValue();
if (value instanceof Map) {
if ("schema".equals(key)) {
return value;
}
return configDesensitization((Map<String, Object>) value);
} else if (value instanceof List) {
return ((List<?>) value)
.stream()
.map(
v -> {
if (v instanceof Map) {
return configDesensitization(
(Map<
String,
Object>)
v);
}
HashMap::new,
(m, p) -> {
String key = p.getKey();
Object value = p.getValue();
if (Arrays.asList(DEFAULT_SENSITIVE_KEYWORDS)
.contains(key.toLowerCase())) {
m.put(key, "******");
} else {
if (value instanceof Map<?, ?>) {
m.put(key, configDesensitization((Map<String, Object>) value));
} else if (value instanceof List<?>) {
List<?> listValue = (List<?>) value;
List<Object> newList =
listValue.stream()
.map(
v -> {
if (v instanceof Map<?, ?>) {
return configDesensitization(
(Map<String, Object>)
v);
} else {
return v;
})
.collect(Collectors.toList());
}
return value;
}));
}
})
.collect(Collectors.toList());
m.put(key, newList);
} else {
m.put(key, value);
}
}
},
HashMap::putAll);
}

public static Config of(
Expand Down Expand Up @@ -185,9 +186,7 @@ private static Config backfillUserVariables(Config config, List<String> variable
return config;
}

public static String mapToString(Map<String, Object> configMap, boolean isJson) {
ConfigRenderOptions configRenderOptions =
ConfigRenderOptions.concise().setFormatted(true).setJson(isJson);
public static String mapToString(Map<String, Object> configMap) {
ConfigParseOptions configParseOptions =
ConfigParseOptions.defaults().setSyntax(ConfigSyntax.JSON);
Config config =
Expand All @@ -196,6 +195,6 @@ public static String mapToString(Map<String, Object> configMap, boolean isJson)
.resolveWith(
ConfigFactory.systemProperties(),
ConfigResolveOptions.defaults().setAllowUnresolved(true));
return config.root().render(configRenderOptions);
return config.root().render(CONFIG_RENDER_OPTIONS);
}
}
27 changes: 24 additions & 3 deletions ...l-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -87,8 +87,8 @@ public void testUsePrivacyHandlerHocon() throws URISyntaxException {
config.getConfigList("source").get(0).getString("username"), "******");
Assertions.assertEquals(
config.getConfigList("source").get(0).getString("password"), "******");
String conf = ConfigBuilder.mapToString(config.root().unwrapped(), false);
Assertions.assertTrue(conf.contains("username=\"******\""));
String conf = ConfigBuilder.mapToString(config.root().unwrapped());
Assertions.assertTrue(conf.contains("\"password\" : \"******\""));
}

@Test
Expand All @@ -107,10 +107,31 @@ public void testUsePrivacyHandlerJson() throws URISyntaxException {
config.getConfigList("source").get(0).getString("username"), "******");
Assertions.assertEquals(
config.getConfigList("source").get(0).getString("password"), "******");
String json = ConfigBuilder.mapToString(config.root().unwrapped(), true);
String json = ConfigBuilder.mapToString(config.root().unwrapped());
Assertions.assertTrue(json.contains("\"password\" : \"******\""));
}

@Test
public void testConfNull() throws URISyntaxException {
URL resource = ConfigShadeTest.class.getResource("/config.shade_caseNull.conf");
Assertions.assertNotNull(resource);
Config config = ConfigBuilder.of(Paths.get(resource.toURI()), Lists.newArrayList());
config =
ConfigFactory.parseMap(
ConfigBuilder.configDesensitization(config.root().unwrapped()))
.resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true))
.resolveWith(
ConfigFactory.systemProperties(),
ConfigResolveOptions.defaults().setAllowUnresolved(true));
Assertions.assertEquals(
config.getConfigList("source").get(0).getString("username"), "******");
Assertions.assertEquals(
config.getConfigList("source").get(0).getString("password"), "******");
String conf = ConfigBuilder.mapToString(config.root().unwrapped());
Assertions.assertTrue(conf.contains("\"password\" : \"******\""));
Assertions.assertTrue(conf.contains("\"test\" : null"));
}

@Test
public void testVariableReplacement() throws URISyntaxException {
String jobName = "seatunnel variable test job";
Expand Down
63 changes: 63 additions & 0 deletions seatunnel-core/seatunnel-core-starter/src/test/resources/config.shade_caseNull.conf
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

env {
parallelism = 1
shade.identifier = "base64"
}

source {
MySQL-CDC {
schema {
fields {
name = string
age = int
sex = boolean
}
}
result_table_name = "fake"
parallelism = 1
server-id = 5656
port = 56725
hostname = "127.0.0.1"
username = "c2VhdHVubmVs"
password = "c2VhdHVubmVsX3Bhc3N3b3Jk"
database-name = "inventory_vwyw0n"
table-name = "products"
base-url = "jdbc:mysql://localhost:56725"
test = null
}
}

transform {
}

sink {
# choose stdout output plugin to output data to console
Clickhouse {
host = "localhost:8123"
database = "default"
table = "fake_all"
username = "c2VhdHVubmVs"
password = "c2VhdHVubmVsX3Bhc3N3b3Jk"

# cdc options
primary_key = "id"
support_upsert = true
test = null
}
}

0 comments on commit 2f31404

Please sign in to comment.