[hotfix][core] Check the cpu and heap memory to be positive when buil…

…ding ResourceSpec
akalash · Jul 5, 2021 · bc5f36d · bc5f36d
1 parent d0846bb
commit bc5f36d
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
@@ -26,6 +26,7 @@
 import javax.annotation.Nullable;
 
 import java.io.Serializable;
+import java.math.BigDecimal;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -70,7 +71,13 @@ public final class ResourceSpec implements Serializable {
     public static final ResourceSpec DEFAULT = UNKNOWN;
 
     /** A ResourceSpec that indicates zero amount of resources. */
-    public static final ResourceSpec ZERO = ResourceSpec.newBuilder(0.0, 0).build();
+    public static final ResourceSpec ZERO =
+            new ResourceSpec(
+                    new CPUResource(0.0),
+                    MemorySize.ZERO,
+                    MemorySize.ZERO,
+                    MemorySize.ZERO,
+                    Collections.emptyMap());
 
     /** How many cpu cores are needed. Can be null only if it is unknown. */
     @Nullable private final CPUResource cpuCores;
@@ -397,6 +404,8 @@ public Builder setExtendedResources(Collection<ExternalResource> extendedResourc
         }
 
         public ResourceSpec build() {
+            checkArgument(cpuCores.getValue().compareTo(BigDecimal.ZERO) > 0);
+            checkArgument(taskHeapMemory.compareTo(MemorySize.ZERO) > 0);
             return new ResourceSpec(
                     cpuCores, taskHeapMemory, taskOffHeapMemory, managedMemory, extendedResources);
         }

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -269,7 +269,7 @@ public void testJobSubmission() throws Exception {
      */
     @Test
     public void testJobSubmissionWithPartialResourceConfigured() throws Exception {
-        ResourceSpec resourceSpec = ResourceSpec.newBuilder(2.0, 0).build();
+        ResourceSpec resourceSpec = ResourceSpec.newBuilder(2.0, 10).build();
 
         final JobVertex firstVertex = new JobVertex("firstVertex");
         firstVertex.setInvokableClass(NoOpInvokable.class);