vsantwana commented on code in PR #1236:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1236#discussion_r4426591262
##########
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) {
+ return Optional.of(e.getMessage());
+ }
return Optional.of(
"TaskManager resource memory must be defined using
`spec.taskManager.resource.memory`"
Review Comment:
This message is now misleading in two cases:
- The spec sets no memory, but `flinkConfiguration` does and conflicts with
another option. For example, `taskmanager.memory.process.size: 2g` with
`taskmanager.memory.network.max: 32m` is reported as "memory must be defined".
The ticket asks for Flink's message instead.
- `resources` sets only cpu and the deprecated `resource.memory` is set.
Ignoring `resource` here is correct, since deploy does the same, but the
message then asks the user to set `spec.taskManager.resource.memory`, which
they already did. `testDeprecatedResourceMemoryIsIgnoredWhenResourcesAreSet`
asserts this message.
Could this message be used only when no memory option is set anywhere, with
Flink's message otherwise? For the second case, it could say that `resource` is
ignored once `resources` is set. The JobManager message above has the same two
cases.
##########
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) {
+ return Optional.of(e.getMessage());
Review Comment:
Flink's message alone can be hard to connect to the spec. With
`resources.requests.memory: 512Mi` it says "... exceed configured Total Flink
Memory (64.000mb ...)", an option the user never set. With `network.max: 32m`
it names `taskmanager.memory.network.fraction`. Could the message say where the
process size came from? For example: "TaskManager memory from
spec.taskManager.resources (512Mi) is inconsistent with flinkConfiguration:
<Flink's message>".
##########
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:
When the spec sets no memory, this branch still doesn't match deploy. On
deploy, Flink always interprets the legacy heap option as
`TOTAL_PROCESS_MEMORY`. I ran both with Flink 1.20.4 and no spec memory:
- `jobmanager.heap.size: 100m` passes validation but fails on deploy ("Sum
of configured JVM Metaspace ... and JVM Overhead ..."). That's the same failure
this PR fixes, just on the other branch.
- `jobmanager.heap.size: 2g` with `jobmanager.memory.process.size: 1g` is
rejected here but deploys fine.
The TaskManager side differs in the same way. Deploy first calls
`TaskExecutorProcessUtils#getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption`,
so `taskmanager.heap.size: 2g` on its own deploys fine but is rejected here.
I know that this predates this PR, but these lines are being rewritten
anyway. Would it be simpler to call `new
KubernetesClusterClientFactory().getClusterSpecification(conf)` for both
components, as deploy and `FlinkUtils#calculateClusterMemoryUsage` do? Then the
validator matches deploy on both branches, and the comment above becomes true
for both.
--
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]
