You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2023/03/28 11:39:30 UTC
[flink] 04/09: [hotfix] Refactor JRR#findParallelism
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0f1536c8a6f66110c50ef57cb88ca5c543aa2104
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Mar 27 12:55:11 2023 +0200
[hotfix] Refactor JRR#findParallelism
Fail if the vertex could not be found.
---
.../flink/runtime/jobgraph/JobResourceRequirements.java | 11 +++++++----
1 file changed, 7 insertions(+), 4 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobResourceRequirements.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobResourceRequirements.java
index 47d99709a93..b282ddcb018 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobResourceRequirements.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobResourceRequirements.java
@@ -118,7 +118,7 @@ public class JobResourceRequirements implements Serializable {
Optional.ofNullable(maxParallelismPerVertex.get(jobVertexId));
if (maybeMaxParallelism.isPresent()) {
final JobVertexResourceRequirements.Parallelism requestedParallelism =
- jobResourceRequirements.findParallelism(jobVertexId).get();
+ jobResourceRequirements.getParallelism(jobVertexId);
int lowerBound =
requestedParallelism.getLowerBound() == -1
? 1
@@ -197,10 +197,13 @@ public class JobResourceRequirements implements Serializable {
Collections.unmodifiableMap(new HashMap<>(checkNotNull(vertexResources)));
}
- public Optional<JobVertexResourceRequirements.Parallelism> findParallelism(
- JobVertexID jobVertexId) {
+ public JobVertexResourceRequirements.Parallelism getParallelism(JobVertexID jobVertexId) {
return Optional.ofNullable(vertexResources.get(jobVertexId))
- .map(JobVertexResourceRequirements::getParallelism);
+ .map(JobVertexResourceRequirements::getParallelism)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "No requirement set for vertex " + jobVertexId));
}
public Set<JobVertexID> getJobVertices() {