Dennis-Mircea commented on code in PR #1236:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1236#discussion_r4426529848
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java:
##########
@@ -328,37 +320,44 @@ private Optional<String>
validateSessionJobJarURI(FlinkSessionJob sessionJob) {
}
private Optional<String> validateJmSpec(JobManagerSpec jmSpec, Map<String,
String> confMap) {
- Configuration conf = Configuration.fromMap(confMap);
- // Either the resource memory or the resource requirements memory must
be defined.
- boolean jmMemoryDefined =
- jmSpec != null
- && (isMemoryDefined(jmSpec.getResource())
- || isMemoryDefined(jmSpec.getResources()));
- Optional<String> jmMemoryValidation =
- jmMemoryDefined ? Optional.empty() :
validateJmMemoryConfig(conf);
-
if (jmSpec == null) {
- return jmMemoryValidation;
+ return validateJmMemoryConfig(null, confMap);
}
return firstPresent(
- jmMemoryValidation,
- validateResources("JobManager", jmSpec.getResource()),
- validateResourceRequirements("JobManager",
jmSpec.getResources()),
- validateJmReplicas(jmSpec.getReplicas(), confMap));
- }
-
- private Optional<String> validateJmMemoryConfig(Configuration conf) {
+ validateResources("JobManager", jmSpec.getResource()),
+ validateResourceRequirements("JobManager",
jmSpec.getResources()),
+ validateJmReplicas(jmSpec.getReplicas(), confMap))
+ .or(() -> validateJmMemoryConfig(jmSpec, confMap));
+ }
+
+ private Optional<String> validateJmMemoryConfig(
+ @Nullable JobManagerSpec jmSpec, Map<String, String> confMap) {
+ final String processMemory =
+ jmSpec == null
+ ? null
+ : ResourceConfigUtils.getProcessMemory(
+ jmSpec.getResources(), jmSpec.getResource());
+ final Configuration conf = Configuration.fromMap(confMap);
+ if (processMemory != null) {
+ conf.setString(JobManagerOptions.TOTAL_PROCESS_MEMORY.key(),
processMemory);
+ }
try {
+ // As on deploy, the legacy heap option is ignored once the spec
sets the memory.
JobManagerProcessUtils.processSpecFromConfigWithNewOptionToInterpretLegacyHeap(
- conf, JobManagerOptions.JVM_HEAP_MEMORY);
+ conf,
+ processMemory == null
+ ? JobManagerOptions.JVM_HEAP_MEMORY
Review Comment:
By still keeping the `JobManagerOptions.JVM_HEAP_MEMORY` as legacy target
currently creates an inconsistency, because the `flink-clients` for native and
standalone deployments uses `JobManagerOptions.TOTAL_PROCESS_MEMORY` instead
(you can check this here
https://github.com/apache/flink/blob/release-1.20/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java#L44).
This concrete inconsistency is in place just for Flink 1.20 version, and
starting with Flink 2.0 there is no longer a legacy target that is being used.
I’d drop this third parameter to promote non-deprecated keys in the
operator. This also aligns it with the TM branch validation, which doesn’t use
a legacy target.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java:
##########
@@ -328,37 +320,44 @@ private Optional<String>
validateSessionJobJarURI(FlinkSessionJob sessionJob) {
}
private Optional<String> validateJmSpec(JobManagerSpec jmSpec, Map<String,
String> confMap) {
- Configuration conf = Configuration.fromMap(confMap);
- // Either the resource memory or the resource requirements memory must
be defined.
- boolean jmMemoryDefined =
- jmSpec != null
- && (isMemoryDefined(jmSpec.getResource())
- || isMemoryDefined(jmSpec.getResources()));
- Optional<String> jmMemoryValidation =
- jmMemoryDefined ? Optional.empty() :
validateJmMemoryConfig(conf);
-
if (jmSpec == null) {
- return jmMemoryValidation;
+ return validateJmMemoryConfig(null, confMap);
}
return firstPresent(
- jmMemoryValidation,
- validateResources("JobManager", jmSpec.getResource()),
- validateResourceRequirements("JobManager",
jmSpec.getResources()),
- validateJmReplicas(jmSpec.getReplicas(), confMap));
- }
-
- private Optional<String> validateJmMemoryConfig(Configuration conf) {
+ validateResources("JobManager", jmSpec.getResource()),
+ validateResourceRequirements("JobManager",
jmSpec.getResources()),
+ validateJmReplicas(jmSpec.getReplicas(), confMap))
+ .or(() -> validateJmMemoryConfig(jmSpec, confMap));
+ }
+
+ private Optional<String> validateJmMemoryConfig(
+ @Nullable JobManagerSpec jmSpec, Map<String, String> confMap) {
+ final String processMemory =
+ jmSpec == null
+ ? null
+ : ResourceConfigUtils.getProcessMemory(
+ jmSpec.getResources(), jmSpec.getResource());
+ final Configuration conf = Configuration.fromMap(confMap);
+ if (processMemory != null) {
+ conf.setString(JobManagerOptions.TOTAL_PROCESS_MEMORY.key(),
processMemory);
+ }
try {
+ // As on deploy, the legacy heap option is ignored once the spec
sets the memory.
JobManagerProcessUtils.processSpecFromConfigWithNewOptionToInterpretLegacyHeap(
- conf, JobManagerOptions.JVM_HEAP_MEMORY);
+ conf,
+ processMemory == null
+ ? JobManagerOptions.JVM_HEAP_MEMORY
+ : JobManagerOptions.TOTAL_PROCESS_MEMORY);
+ return Optional.empty();
} catch (Exception e) {
+ if (processMemory != null) {
+ return Optional.of(e.getMessage());
+ }
return Optional.of(
"JobManager resource memory must be defined using
`spec.jobManager.resource.memory`"
+ " or
`spec.jobManager.resources.requests.memory`");
Review Comment:
This branching is a bit unclear to me, and I think the messages would make
more sense if they were swapped. When `processMemory != null`,
`spec.jobManager` has actually been defined, yet the user gets a generic Flink
core message. Meanwhile, when the user only configures the memory properties
under `spec.flinkConfiguration`, they get the old error message. This is quite
confusing, so I’d suggest restructuring it like this:
```java
return Optional.of(
processMemory == null
? e.getMessage()
: e.getMessage()
+ " (jobmanager.memory.process.size is set from
the spec.jobManager memory)");
```
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java:
##########
@@ -374,35 +373,39 @@ private Optional<String> validateJmReplicas(int replicas,
Map<String, String> co
}
private Optional<String> validateTmSpec(TaskManagerSpec tmSpec,
Map<String, String> confMap) {
- Configuration conf = Configuration.fromMap(confMap);
-
- var tmMemoryDefined =
- tmSpec != null
- && (isMemoryDefined(tmSpec.getResource())
- || isMemoryDefined(tmSpec.getResources()));
- Optional<String> tmMemoryConfigValidation =
- tmMemoryDefined ? Optional.empty() :
validateTmMemoryConfig(conf);
-
if (tmSpec == null) {
- return tmMemoryConfigValidation;
+ return validateTmMemoryConfig(null, confMap);
}
return firstPresent(
- tmMemoryConfigValidation,
- validateResources("TaskManager", tmSpec.getResource()),
- validateResourceRequirements("TaskManager",
tmSpec.getResources()),
- validateTmReplicas(tmSpec));
- }
-
- private Optional<String> validateTmMemoryConfig(Configuration conf) {
+ validateResources("TaskManager", tmSpec.getResource()),
+ validateResourceRequirements("TaskManager",
tmSpec.getResources()),
+ validateTmReplicas(tmSpec))
+ .or(() -> validateTmMemoryConfig(tmSpec, confMap));
+ }
+
+ private Optional<String> validateTmMemoryConfig(
+ @Nullable TaskManagerSpec tmSpec, Map<String, String> confMap) {
+ final String processMemory =
+ tmSpec == null
+ ? null
+ : ResourceConfigUtils.getProcessMemory(
+ tmSpec.getResources(), tmSpec.getResource());
+ final Configuration conf = Configuration.fromMap(confMap);
+ if (processMemory != null) {
+ conf.setString(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
processMemory);
Review Comment:
Same here.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java:
##########
@@ -374,35 +373,39 @@ private Optional<String> validateJmReplicas(int replicas,
Map<String, String> co
}
private Optional<String> validateTmSpec(TaskManagerSpec tmSpec,
Map<String, String> confMap) {
- Configuration conf = Configuration.fromMap(confMap);
-
- var tmMemoryDefined =
- tmSpec != null
- && (isMemoryDefined(tmSpec.getResource())
- || isMemoryDefined(tmSpec.getResources()));
- Optional<String> tmMemoryConfigValidation =
- tmMemoryDefined ? Optional.empty() :
validateTmMemoryConfig(conf);
-
if (tmSpec == null) {
- return tmMemoryConfigValidation;
+ return validateTmMemoryConfig(null, confMap);
}
return firstPresent(
- tmMemoryConfigValidation,
- validateResources("TaskManager", tmSpec.getResource()),
- validateResourceRequirements("TaskManager",
tmSpec.getResources()),
- validateTmReplicas(tmSpec));
- }
-
- private Optional<String> validateTmMemoryConfig(Configuration conf) {
+ validateResources("TaskManager", tmSpec.getResource()),
+ validateResourceRequirements("TaskManager",
tmSpec.getResources()),
+ validateTmReplicas(tmSpec))
+ .or(() -> validateTmMemoryConfig(tmSpec, confMap));
+ }
+
+ private Optional<String> validateTmMemoryConfig(
+ @Nullable TaskManagerSpec tmSpec, Map<String, String> confMap) {
+ final String processMemory =
+ tmSpec == null
+ ? null
+ : ResourceConfigUtils.getProcessMemory(
+ tmSpec.getResources(), tmSpec.getResource());
+ final Configuration conf = Configuration.fromMap(confMap);
+ if (processMemory != null) {
+ conf.setString(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
processMemory);
+ }
try {
TaskExecutorProcessUtils.processSpecFromConfig(conf);
+ return Optional.empty();
} catch (Exception e) {
+ if (processMemory != null) {
Review Comment:
Same here on the error message restructuring.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java:
##########
@@ -328,37 +320,44 @@ private Optional<String>
validateSessionJobJarURI(FlinkSessionJob sessionJob) {
}
private Optional<String> validateJmSpec(JobManagerSpec jmSpec, Map<String,
String> confMap) {
- Configuration conf = Configuration.fromMap(confMap);
- // Either the resource memory or the resource requirements memory must
be defined.
- boolean jmMemoryDefined =
- jmSpec != null
- && (isMemoryDefined(jmSpec.getResource())
- || isMemoryDefined(jmSpec.getResources()));
- Optional<String> jmMemoryValidation =
- jmMemoryDefined ? Optional.empty() :
validateJmMemoryConfig(conf);
-
if (jmSpec == null) {
- return jmMemoryValidation;
+ return validateJmMemoryConfig(null, confMap);
}
return firstPresent(
- jmMemoryValidation,
- validateResources("JobManager", jmSpec.getResource()),
- validateResourceRequirements("JobManager",
jmSpec.getResources()),
- validateJmReplicas(jmSpec.getReplicas(), confMap));
- }
-
- private Optional<String> validateJmMemoryConfig(Configuration conf) {
+ validateResources("JobManager", jmSpec.getResource()),
+ validateResourceRequirements("JobManager",
jmSpec.getResources()),
+ validateJmReplicas(jmSpec.getReplicas(), confMap))
+ .or(() -> validateJmMemoryConfig(jmSpec, confMap));
+ }
+
+ private Optional<String> validateJmMemoryConfig(
+ @Nullable JobManagerSpec jmSpec, Map<String, String> confMap) {
+ final String processMemory =
+ jmSpec == null
+ ? null
+ : ResourceConfigUtils.getProcessMemory(
+ jmSpec.getResources(), jmSpec.getResource());
+ final Configuration conf = Configuration.fromMap(confMap);
+ if (processMemory != null) {
+ conf.setString(JobManagerOptions.TOTAL_PROCESS_MEMORY.key(),
processMemory);
Review Comment:
This is redundant. The FlinkConfigBuilder already adds this key in the
deployed config when the memory is specified under `spec.jobManager.resources`
or `spec.jobManager.resource`.
--
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]
