[flink] branch release-1.11 updated: [FLINK-17503][runtime] [logs] Refactored log output.

[tr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.11 by this push:
     new fab025d  [FLINK-17503][runtime] [logs] Refactored log output.
fab025d is described below

commit fab025d84a3b0ba24d89888e977dd05066e6dbcb
Author: Matthias Pohl <ma...@ververica.com>
AuthorDate: Fri Aug 7 16:01:36 2020 +0200

    [FLINK-17503][runtime] [logs] Refactored log output.
    
    The log output was cleaned up:
     - Only the key of the option is logged out instead of the whole
       instance's toString() method.
     - A new utility method was introduce that adapts the log output in a
       way that an adapted extension is used for maximum values instead of
       logging the actual max value.
    
    This closes #13086.
---
 .../taskexecutor/TaskExecutorResourceUtils.java    | 28 +++++++++++++++++-----
 1 file changed, 22 insertions(+), 6 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java
index 0b3868d..8653dbe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorResourceUtils.java
@@ -132,9 +132,10 @@ public class TaskExecutorResourceUtils {
 	public static Configuration adjustForLocalExecution(Configuration config) {
 		UNUSED_CONFIG_OPTIONS.forEach(option -> warnOptionHasNoEffectIfSet(config, option));
 
-		setConfigOptionToDefaultIfNotSet(config, TaskManagerOptions.CPU_CORES, Double.MAX_VALUE);
-		setConfigOptionToDefaultIfNotSet(config, TaskManagerOptions.TASK_HEAP_MEMORY, MemorySize.MAX_VALUE);
-		setConfigOptionToDefaultIfNotSet(config, TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.MAX_VALUE);
+		setConfigOptionToPassedMaxIfNotSet(config, TaskManagerOptions.CPU_CORES, Double.MAX_VALUE);
+		setConfigOptionToPassedMaxIfNotSet(config, TaskManagerOptions.TASK_HEAP_MEMORY, MemorySize.MAX_VALUE);
+		setConfigOptionToPassedMaxIfNotSet(config, TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.MAX_VALUE);
+
 		adjustNetworkMemoryForLocalExecution(config);
 		setConfigOptionToDefaultIfNotSet(config, TaskManagerOptions.MANAGED_MEMORY_SIZE, DEFAULT_MANAGED_MEMORY_SIZE);
 
@@ -168,11 +169,26 @@ public class TaskExecutorResourceUtils {
 			Configuration config,
 			ConfigOption<T> option,
 			T defaultValue) {
+		setConfigOptionToDefaultIfNotSet(config, option, defaultValue, "its default value " + defaultValue);
+	}
+
+	private static <T> void setConfigOptionToPassedMaxIfNotSet(
+			Configuration config,
+			ConfigOption<T> option,
+			T maxValue) {
+		setConfigOptionToDefaultIfNotSet(config, option, maxValue, "the maximal possible value");
+	}
+
+	private static <T> void setConfigOptionToDefaultIfNotSet(
+			Configuration config,
+			ConfigOption<T> option,
+			T defaultValue,
+			String defaultValueLogExt) {
 		if (!config.contains(option)) {
 			LOG.info(
-				"The configuration option {} required for local execution is not set, setting it to its default value {}",
-				option,
-				defaultValue);
+				"The configuration option {} required for local execution is not set, setting it to {}.",
+				option.key(),
+				defaultValueLogExt);
 			config.set(option, defaultValue);
 		}
 	}