Revert "[FLINK-10848][YARN] properly remove YARN ContainerRequest upon...

Revert "[FLINK-10848][YARN] properly remove YARN ContainerRequest upon container allocation success"

This reverts commit e26d90fc.

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java

@@ -165,8 +165,6 @@ public abstract class YarnTestBase extends TestLogger {
 		YARN_CONFIGURATION.setInt(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC, 3600);
 		YARN_CONFIGURATION.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, false);
 		YARN_CONFIGURATION.setInt(YarnConfiguration.NM_VCORES, 666); // memory is overwritten in the MiniYARNCluster.
-		YARN_CONFIGURATION.set("yarn.scheduler.capacity.resource-calculator",
-				"org.apache.hadoop.yarn.util.resource.DominantResourceCalculator");
 		// so we have to change the number of cores for testing.
 		YARN_CONFIGURATION.setInt(YarnConfiguration.RM_AM_EXPIRY_INTERVAL_MS, 20000); // 20 seconds expiry (to ensure we properly heartbeat with YARN).
 	}

flink-yarn/src/main/java/org/apache/flink/yarn/YarnFlinkResourceManager.java

@@ -438,8 +438,6 @@ public class YarnFlinkResourceManager extends FlinkResourceManager<RegisteredYar
 			numPendingContainerRequests = Math.max(0, numPendingContainerRequests - 1);
 			LOG.info("Received new container: {} - Remaining pending container requests: {}",
 				container.getId(), numPendingContainerRequests);
-			resourceManagerClient.removeContainerRequest(new AMRMClient.ContainerRequest(
-					container.getResource(), null, null, container.getPriority()));
 
 			// decide whether to return the container, or whether to start a TaskManager
 			if (numRegistered + containersInLaunch.size() < numRequired) {

flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java

@@ -361,8 +361,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
 					"Received new container: {} - Remaining pending container requests: {}",
 					container.getId(),
 					numPendingContainerRequests);
-				resourceManagerClient.removeContainerRequest(new AMRMClient.ContainerRequest(
-						container.getResource(), null, null, container.getPriority()));
+
 				if (numPendingContainerRequests > 0) {
 					numPendingContainerRequests--;
 

flink-yarn/src/test/java/org/apache/flink/yarn/YarnFlinkResourceManagerTest.java

@@ -43,8 +43,6 @@ import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
 import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.Priority;
-import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.client.api.AMRMClient;
 import org.apache.hadoop.yarn.client.api.NMClient;
 import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
@@ -71,11 +69,8 @@ import scala.concurrent.duration.Deadline;
 import scala.concurrent.duration.FiniteDuration;
 
 import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 /**
@@ -130,8 +125,6 @@ public class YarnFlinkResourceManagerTest extends TestLogger {
 							1),
 						i));
 				when(mockContainer.getNodeId()).thenReturn(NodeId.newInstance("container", 1234));
-				when(mockContainer.getResource()).thenReturn(Resource.newInstance(200, 1));
-				when(mockContainer.getPriority()).thenReturn(Priority.UNDEFINED);
 				containerList.add(mockContainer);
 			}
 
@@ -240,8 +233,6 @@ public class YarnFlinkResourceManagerTest extends TestLogger {
 
 				int numberOfRegisteredResources = (Integer) Await.result(numberOfRegisteredResourcesFuture, deadline.timeLeft());
 
-				verify(resourceManagerClient, times(numInitialTaskManagers)).removeContainerRequest(
-						any(AMRMClient.ContainerRequest.class));
 				assertEquals(numInitialTaskManagers, numberOfRegisteredResources);
 			} finally {
 				if (resourceManager != null) {

flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java

@@ -401,8 +401,6 @@ public class YarnResourceManagerTest extends TestLogger {
 
 				resourceManager.onContainersAllocated(ImmutableList.of(testingContainer));
 				verify(mockResourceManagerClient).addContainerRequest(any(AMRMClient.ContainerRequest.class));
-				verify(mockResourceManagerClient).removeContainerRequest(
-						any(AMRMClient.ContainerRequest.class));
 				verify(mockNMClient).startContainer(eq(testingContainer), any(ContainerLaunchContext.class));
 
 				// Remote task executor registers with YarnResourceManager.
@@ -498,8 +496,6 @@ public class YarnResourceManagerTest extends TestLogger {
 
 				resourceManager.onContainersAllocated(ImmutableList.of(testingContainer));
 				verify(mockResourceManagerClient).addContainerRequest(any(AMRMClient.ContainerRequest.class));
-				verify(mockResourceManagerClient).removeContainerRequest(
-						any(AMRMClient.ContainerRequest.class));
 				verify(mockNMClient).startContainer(eq(testingContainer), any(ContainerLaunchContext.class));
 
 				// Callback from YARN when container is Completed, pending request can not be fulfilled by pending