You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/08/31 10:12:51 UTC
[flink] branch master updated: [FLINK-32817][Yarn] Harnessing Jackson for Secure Serialization of YarnLocalResourceDescriptor
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1d1247d4ae6 [FLINK-32817][Yarn] Harnessing Jackson for Secure Serialization of YarnLocalResourceDescriptor
1d1247d4ae6 is described below
commit 1d1247d4ae6d4313f7d952c4b2d66351314c9432
Author: Xianxun Ye <ye...@gmail.com>
AuthorDate: Fri Aug 25 18:02:54 2023 +0800
[FLINK-32817][Yarn] Harnessing Jackson for Secure Serialization of YarnLocalResourceDescriptor
This closes #23292
---
.../flink/util/jackson/JacksonMapperFactory.java | 7 ++
.../flink/yarn/YarnLocalResourceDescriptor.java | 80 ++++++++++++++--------
.../yarn/YarnLocalResourceDescriptionTest.java | 12 +++-
3 files changed, 68 insertions(+), 31 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/jackson/JacksonMapperFactory.java b/flink-core/src/main/java/org/apache/flink/util/jackson/JacksonMapperFactory.java
index e43451f92b5..e19f4578d74 100644
--- a/flink-core/src/main/java/org/apache/flink/util/jackson/JacksonMapperFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/util/jackson/JacksonMapperFactory.java
@@ -19,6 +19,7 @@ package org.apache.flink.util.jackson;
import org.apache.flink.annotation.Experimental;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper;
@@ -35,6 +36,12 @@ public final class JacksonMapperFactory {
return objectMapper;
}
+ public static ObjectMapper createObjectMapper(JsonFactory jsonFactory) {
+ final ObjectMapper objectMapper = new ObjectMapper(jsonFactory);
+ registerModules(objectMapper);
+ return objectMapper;
+ }
+
public static CsvMapper createCsvMapper() {
final CsvMapper csvMapper = new CsvMapper();
registerModules(csvMapper);
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnLocalResourceDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnLocalResourceDescriptor.java
index 87de5d5684a..16f7ec1d8fc 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnLocalResourceDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnLocalResourceDescriptor.java
@@ -19,16 +19,24 @@
package org.apache.flink.yarn;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactoryBuilder;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Objects;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -38,14 +46,12 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* {@link Utils#createTaskExecutorContext}.
*/
class YarnLocalResourceDescriptor {
+ private static final Logger LOG = LoggerFactory.getLogger(YarnLocalResourceDescriptor.class);
- private static final String STRING_FORMAT =
- "YarnLocalResourceDescriptor{"
- + "key=%s, path=%s, size=%d, modificationTime=%d, visibility=%s, type=%s}";
- private static final Pattern LOCAL_RESOURCE_DESC_FORMAT =
- Pattern.compile(
- "YarnLocalResourceDescriptor\\{"
- + "key=(\\S+), path=(\\S+), size=([\\d]+), modificationTime=([\\d]+), visibility=(\\S+), type=(\\S+)}");
+ private static final ObjectMapper OBJECT_MAPPER =
+ JacksonMapperFactory.createObjectMapper(
+ new JsonFactoryBuilder().quoteChar('\'').build())
+ .enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES);
private final String resourceKey;
private final Path path;
@@ -98,21 +104,33 @@ class YarnLocalResourceDescriptor {
}
static YarnLocalResourceDescriptor fromString(String desc) throws Exception {
- Matcher m = LOCAL_RESOURCE_DESC_FORMAT.matcher(desc);
- boolean mat = m.find();
- if (mat) {
+ try {
+ JsonNode node = OBJECT_MAPPER.readTree(desc);
+ if (!validate(node)) {
+ throw new FlinkException("Error to parse YarnLocalResourceDescriptor from " + desc);
+ }
return new YarnLocalResourceDescriptor(
- m.group(1),
- new Path(m.group(2)),
- Long.parseLong(m.group(3)),
- Long.parseLong(m.group(4)),
- LocalResourceVisibility.valueOf(m.group(5)),
- LocalResourceType.valueOf(m.group(6)));
- } else {
- throw new FlinkException("Error to parse YarnLocalResourceDescriptor from " + desc);
+ node.get("resourceKey").asText(),
+ new Path(node.get("path").asText()),
+ node.get("size").asLong(),
+ node.get("modificationTime").asLong(),
+ LocalResourceVisibility.valueOf(node.get("visibility").asText()),
+ LocalResourceType.valueOf(node.get("resourceType").asText()));
+ } catch (JsonProcessingException e) {
+ throw new FlinkException("Error to parse YarnLocalResourceDescriptor from " + desc, e);
}
}
+ private static boolean validate(JsonNode node) {
+ return !node.isNull()
+ && node.hasNonNull("resourceKey")
+ && node.hasNonNull("path")
+ && node.hasNonNull("size")
+ && node.hasNonNull("modificationTime")
+ && node.hasNonNull("visibility")
+ && node.hasNonNull("resourceType");
+ }
+
static YarnLocalResourceDescriptor fromFileStatus(
final String key,
final FileStatus fileStatus,
@@ -132,14 +150,20 @@ class YarnLocalResourceDescriptor {
@Override
public String toString() {
- return String.format(
- STRING_FORMAT,
- resourceKey,
- path.toString(),
- size,
- modificationTime,
- visibility,
- resourceType);
+ try {
+ ObjectNode node = OBJECT_MAPPER.createObjectNode();
+ node.put("resourceKey", resourceKey);
+ node.put("path", path.toString());
+ node.put("size", size);
+ node.put("modificationTime", modificationTime);
+ node.put("visibility", visibility.toString());
+ node.put("resourceType", resourceType.toString());
+ return OBJECT_MAPPER.writeValueAsString(node);
+ } catch (JsonProcessingException e) {
+ LOG.error("Could not serialize YarnLocalResourceDescriptor to String.", e);
+ throw new RuntimeException(
+ "Could not serialize YarnLocalResourceDescriptor[%s] to String.", e);
+ }
}
@Override
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnLocalResourceDescriptionTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnLocalResourceDescriptionTest.java
index bb4c414dd3e..b301c1ef678 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnLocalResourceDescriptionTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnLocalResourceDescriptionTest.java
@@ -31,8 +31,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link YarnLocalResourceDescriptor}. */
class YarnLocalResourceDescriptionTest {
- private final String key = "flink.jar";
- private final Path path = new Path("hdfs://nn/tmp/flink.jar");
+ private final String key = "fli'nk 2.jar";
+ private final Path path = new Path("hdfs://nn/tmp/fli'nk 2.jar");
private final long size = 100 * 1024 * 1024;
private final long ts = System.currentTimeMillis();
@@ -62,8 +62,14 @@ class YarnLocalResourceDescriptionTest {
void testFromStringMalformed() {
final String desc =
String.format(
- "YarnLocalResourceDescriptor{key=%s path=%s size=%d modTime=%d visibility=%s}",
+ "{'resourceKey':'%s','path':'%s','size':%s,'modificationTime':%s,'visibility':'%s'}",
key, path, size, ts, LocalResourceVisibility.PUBLIC);
+ assertThrows(desc);
+ assertThrows("{}");
+ assertThrows("{");
+ }
+
+ private void assertThrows(final String desc) {
assertThatThrownBy(() -> YarnLocalResourceDescriptor.fromString(desc))
.isInstanceOf(FlinkException.class)
.hasMessageContaining("Error to parse YarnLocalResourceDescriptor from " + desc);