From 7c5448c991b0fad3b28784ec0b99324df948f498 Mon Sep 17 00:00:00 2001
From: Xintong Song <tonysong820@gmail.com>
Date: Fri, 19 Mar 2021 11:48:05 +0800
Subject: [PATCH] [hotfix][core][test] Prioritize configurations over system
 properties for enabling fine-grained resource management.

---
 .../flink/configuration/ClusterOptions.java   | 12 +++++++---
 .../ActiveResourceManagerFactoryTest.java     | 22 +------------------
 2 files changed, 10 insertions(+), 24 deletions(-)

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
index 61b47e405a8f4..cefc55015cd6d 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
@@ -165,9 +165,15 @@ public static boolean isFineGrainedResourceManagementEnabled(Configuration confi
         // TODO We need to bind fine-grained with declarative because in the first step we implement
         // the feature base on the declarative protocol. We would be able to support both protocols
         // and no longer need this binding after FLINK-20838.
-        return isDeclarativeResourceManagementEnabled(configuration)
-                && (configuration.get(ENABLE_FINE_GRAINED_RESOURCE_MANAGEMENT)
-                        || System.getProperties().containsKey("flink.tests.enable-fine-grained"));
+        if (!isDeclarativeResourceManagementEnabled(configuration)) {
+            return false;
+        }
+
+        if (configuration.contains(ENABLE_FINE_GRAINED_RESOURCE_MANAGEMENT)) {
+            return configuration.get(ENABLE_FINE_GRAINED_RESOURCE_MANAGEMENT);
+        } else {
+            return System.getProperties().containsKey("flink.tests.enable-fine-grained");
+        }
     }
 
     /** The mode of how to handle user code attempting to exit JVM. */
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManagerFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManagerFactoryTest.java
index fda827bc64cc8..32483b12982f9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManagerFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManagerFactoryTest.java
@@ -22,20 +22,15 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.MemorySize;
 import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.core.testutils.CommonTestUtils;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServicesConfiguration;
 import org.apache.flink.util.ConfigurationException;
 import org.apache.flink.util.TestLogger;
 
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
 import org.junit.Test;
 
 import javax.annotation.Nullable;
 
-import java.util.Map;
-
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
@@ -47,25 +42,10 @@ public class ActiveResourceManagerFactoryTest extends TestLogger {
     private static final MemorySize TOTAL_FLINK_SIZE = MemorySize.ofMebiBytes(2 * 1024);
     private static final MemorySize TOTAL_PROCESS_SIZE = MemorySize.ofMebiBytes(3 * 1024);
 
-    private static Map<String, String> systemEnv;
-
-    @BeforeClass
-    public static void setupClass() {
-        systemEnv = System.getenv();
-        System.clearProperty("flink.tests.disable-declarative");
-        System.clearProperty("flink.tests.enable-fine-grained");
-    }
-
-    @AfterClass
-    public static void teardownClass() {
-        if (systemEnv != null) {
-            CommonTestUtils.setEnv(systemEnv, true);
-        }
-    }
-
     @Test
     public void testGetEffectiveConfigurationForResourceManagerCoarseGrained() {
         final Configuration config = new Configuration();
+        config.set(ClusterOptions.ENABLE_DECLARATIVE_RESOURCE_MANAGEMENT, false);
         config.set(ClusterOptions.ENABLE_FINE_GRAINED_RESOURCE_MANAGEMENT, false);
         config.set(TaskManagerOptions.TOTAL_FLINK_MEMORY, TOTAL_FLINK_SIZE);
         config.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, TOTAL_PROCESS_SIZE);