[FLINK-21943][core] Remove redundant checkNotNull in Transformation#setResources

Author: plugatarev

flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java

@@ -33,6 +33,7 @@
 import javax.annotation.Nullable;
 
 import java.util.Collections;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -42,7 +43,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
-import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * A {@code Transformation} represents the operation that creates a DataStream. Every DataStream has
@@ -160,7 +160,7 @@ public static int getNewNodeId() {
      * will be shared by all the declaring transformations within a slot according to this weight.
      */
     private final Map<ManagedMemoryUseCase, Integer> managedMemoryOperatorScopeUseCaseWeights =
-            new HashMap<>();
+            new EnumMap<>(ManagedMemoryUseCase.class);
 
     /**
      * This map is a cache that stores transitive predecessors and used in {@code
@@ -306,8 +306,8 @@ public void setMaxParallelism(int maxParallelism) {
      */
     public void setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
         OperatorValidationUtils.validateMinAndPreferredResources(minResources, preferredResources);
-        this.minResources = checkNotNull(minResources);
-        this.preferredResources = checkNotNull(preferredResources);
+        this.minResources = minResources;
+        this.preferredResources = preferredResources;
     }
 
     /**

flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.dag;
 
+import org.apache.flink.api.common.operators.ResourceSpec;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.core.memory.ManagedMemoryUseCase;
 import org.apache.flink.core.testutils.CheckedThread;
@@ -134,6 +135,16 @@ void testDeclareManagedMemorySlotScopeUseCaseFailWrongScope() {
                 .isInstanceOf(IllegalArgumentException.class);
     }
 
+    @Test
+    void testSetResourcesUseCaseFailNullResources() {
+        ResourceSpec resourceSpec = ResourceSpec.newBuilder(1.0, 100).build();
+
+        assertThatThrownBy(() -> transformation.setResources(null, resourceSpec))
+                .isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> transformation.setResources(resourceSpec, null))
+                .isInstanceOf(NullPointerException.class);
+    }
+
     /** A test implementation of {@link Transformation}. */
     private static class TestTransformation<T> extends Transformation<T> {