[GitHub] flink pull request #5695: [FLINK-8704][tests] Port PartialConsumerPipelinedR...

[zentol <gi...@git.apache.org>]

Github user zentol commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5695#discussion_r178752435
  
    --- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java ---
    @@ -32,41 +31,51 @@
     import org.apache.flink.runtime.jobgraph.JobVertex;
     import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
     import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
    -import org.apache.flink.runtime.testingUtils.TestingCluster;
    +import org.apache.flink.runtime.minicluster.MiniCluster;
    +import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
     import org.apache.flink.runtime.testingUtils.TestingUtils;
    +import org.apache.flink.testutils.category.Flip6;
     import org.apache.flink.util.TestLogger;
     
     import org.junit.AfterClass;
     import org.junit.BeforeClass;
     import org.junit.Test;
    +import org.junit.experimental.categories.Category;
     
    +@Category(Flip6.class)
     public class PartialConsumePipelinedResultTest extends TestLogger {
     
     	// Test configuration
    -	private final static int NUMBER_OF_TMS = 1;
    -	private final static int NUMBER_OF_SLOTS_PER_TM = 1;
    -	private final static int PARALLELISM = NUMBER_OF_TMS * NUMBER_OF_SLOTS_PER_TM;
    +	private static final int NUMBER_OF_TMS = 1;
    +	private static final int NUMBER_OF_SLOTS_PER_TM = 1;
    +	private static final int PARALLELISM = NUMBER_OF_TMS * NUMBER_OF_SLOTS_PER_TM;
     
    -	private final static int NUMBER_OF_NETWORK_BUFFERS = 128;
    +	private static final int NUMBER_OF_NETWORK_BUFFERS = 128;
     
    -	private static TestingCluster flink;
    +	private static MiniCluster flink;
     
     	@BeforeClass
     	public static void setUp() throws Exception {
     		final Configuration config = new Configuration();
    -		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUMBER_OF_TMS);
    -		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUMBER_OF_SLOTS_PER_TM);
     		config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
     		config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, NUMBER_OF_NETWORK_BUFFERS);
     
    -		flink = new TestingCluster(config, true);
    +		final MiniClusterConfiguration miniClusterConfiguration = new MiniClusterConfiguration.Builder()
    +			.setConfiguration(config)
    +			.setNumTaskManagers(NUMBER_OF_TMS)
    +			.setNumSlotsPerTaskManager(NUMBER_OF_SLOTS_PER_TM)
    +			.build();
    +
    +		flink = new MiniCluster(miniClusterConfiguration);
     
     		flink.start();
     	}
     
     	@AfterClass
     	public static void tearDown() throws Exception {
    -		flink.stop();
    +		if (flink != null) {
    +			flink.close();
    +		}
    --- End diff --
    
    technically it is required in the legacy test as well, but i just copied it as i did for all other tests.


---