kkdoon commented on code in PR #26009:
URL: https://github.com/apache/beam/pull/26009#discussion_r1177145774
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java:
##########
@@ -140,6 +150,9 @@ static ExecutionEnvironment createBatchExecutionEnvironment(
applyLatencyTrackingInterval(flinkBatchEnv.getConfig(), options);
+ configureWebUIOptions(
+ flinkBatchEnv.getConfig(),
options.as(org.apache.beam.sdk.options.PipelineOptions.class));
Review Comment:
good catch! will import it instead.
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java:
##########
@@ -245,9 +258,63 @@ public static StreamExecutionEnvironment
createStreamExecutionEnvironment(
configureStateBackend(options, flinkStreamEnv);
+ configureWebUIOptions(
+ flinkStreamEnv.getConfig(),
options.as(org.apache.beam.sdk.options.PipelineOptions.class));
+
return flinkStreamEnv;
}
+ private static void configureWebUIOptions(
+ ExecutionConfig config, org.apache.beam.sdk.options.PipelineOptions
options) {
+ SerializablePipelineOptions serializablePipelineOptions =
+ new SerializablePipelineOptions(options);
+ String optionsAsString = serializablePipelineOptions.toString();
+
+ try {
+ JsonNode node = mapper.readTree(optionsAsString);
+ JsonNode optionsNode = node.get("options");
+ Map<String, String> output =
+ Streams.stream(optionsNode.fields())
+ .filter(
+ entry ->
+ !entry.getValue().asText().isEmpty()
+ && !"null".equals(entry.getValue().asText()))
Review Comment:
good point, i can use `isNull` method in JsonNode class to detect null
scenarios.
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java:
##########
@@ -245,9 +258,63 @@ public static StreamExecutionEnvironment
createStreamExecutionEnvironment(
configureStateBackend(options, flinkStreamEnv);
+ configureWebUIOptions(
+ flinkStreamEnv.getConfig(),
options.as(org.apache.beam.sdk.options.PipelineOptions.class));
+
return flinkStreamEnv;
}
+ private static void configureWebUIOptions(
+ ExecutionConfig config, org.apache.beam.sdk.options.PipelineOptions
options) {
+ SerializablePipelineOptions serializablePipelineOptions =
+ new SerializablePipelineOptions(options);
+ String optionsAsString = serializablePipelineOptions.toString();
+
+ try {
+ JsonNode node = mapper.readTree(optionsAsString);
+ JsonNode optionsNode = node.get("options");
+ Map<String, String> output =
+ Streams.stream(optionsNode.fields())
+ .filter(
+ entry ->
+ !entry.getValue().asText().isEmpty()
Review Comment:
makes sense, i will remove this check. `IsNull` is sufficient to check
whether the field is set or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
