/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.runtime.clusterframework;

import org.apache.flink.configuration.Configuration;
import org.junit.Test;

import static org.apache.flink.configuration.ConfigConstants.TASK_MANAGER_MEMORY_OFF_HEAP_KEY;
import static org.apache.flink.configuration.TaskManagerOptions.MANAGED_MEMORY_SIZE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

public class ContaineredTaskManagerParametersTest {
	private static final long CONTAINER_MEMORY = 8192;

	@Test
	public void testDefaultOffHeapMemory() {
		Configuration conf = new Configuration();
		ContaineredTaskManagerParameters params =
			ContaineredTaskManagerParameters.create(conf, CONTAINER_MEMORY, 1);
		assertEquals(-1, params.taskManagerDirectMemoryLimitMB());
	}

	@Test
	public void testTotalMemoryDoesNotExceedContainerMemory() {
		Configuration conf = new Configuration();
		conf.setBoolean(MANAGED_MEMORY_SIZE.key(), true);
		ContaineredTaskManagerParameters params =
			ContaineredTaskManagerParameters.create(conf, CONTAINER_MEMORY, 1);
		assertTrue(params.taskManagerHeapSizeMB() +
			params.taskManagerDirectMemoryLimitMB() <= CONTAINER_MEMORY);
	}
}

@@ -920,115 +920,61 @@
 org.
-junit.Test
+apache.flink.util.TestLogger
 ;%0A
-%0A
 import 
-static org.apache.flink.configuration.ConfigConstants.TASK_MANAGER_MEMORY_OFF_HEAP_KEY;
+org.junit.Test;%0A
 %0Aimp
@@ -1035,34 +1035,30 @@
 ptions.M
-ANAGED_MEMORY_SIZE
+EMORY_OFF_HEAP
 ;%0Aimport
@@ -1190,16 +1190,35 @@
 ersTest 
+extends TestLogger 
 %7B%0A%09priva
@@ -1261,19 +1261,96 @@
 Y = 8192
+L
 ;%0A%0A
+%09/**%0A%09 * This tests that per default the off heap memory is set to -1.%0A%09 */%0A
 %09@Test%0A%09
@@ -1369,15 +1369,8 @@
 test
-Default
 OffH
@@ -1374,24 +1374,48 @@
 ffHeapMemory
+WithDefaultConfiguration
 () %7B%0A%09%09Confi
@@ -1451,16 +1451,17 @@
 tion();%0A
+%0A
 %09%09Contai
@@ -1584,16 +1584,17 @@
 quals(-1
+L
 , params
@@ -1633,16 +1633,151 @@
 ));%0A%09%7D%0A%0A
+%09/**%0A%09 * This tests that when using off heap memory the sum of on and off heap memory does not exceeds the container%0A%09 * maximum.%0A%09 */%0A
 %09@Test%0A%09
@@ -1903,32 +1903,22 @@
 an(M
-ANAGED_MEMORY_SIZE.key()
+EMORY_OFF_HEAP
 , tr
@@ -1922,16 +1922,17 @@
  true);%0A
+%0A
 %09%09Contai
@@ -2030,32 +2030,94 @@
 NER_MEMORY, 1);%0A
+%0A%09%09assertTrue(params.taskManagerDirectMemoryLimitMB() %3E 0L);%0A%0A
 %09%09assertTrue(par

Fix ContaineredTaskManagerParametersTest to properly test off heap settings