Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5737#discussion_r176208188
--- Diff:
flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java
---
@@ -45,84 +47,87 @@
* b) are not compatible with existing accumulator.
*/
public class AccumulatorErrorITCase extends TestLogger {
-
- private static LocalFlinkMiniCluster cluster;
-
- private static ExecutionEnvironment env;
-
- @BeforeClass
- public static void startCluster() {
+ private static final String FAULTY_CLONE_ACCUMULATOR = "faulty-clone";
+ private static final String FAULTY_MERGE_ACCUMULATOR = "faulty-merge";
+ private static final String INCOMPATIBLE_ACCUMULATORS_NAME =
"incompatible-accumulators";
+
+ @ClassRule
+ public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new
MiniClusterResource(
+ new MiniClusterResource.MiniClusterResourceConfiguration(
+ getConfiguration(),
+ 2,
+ 3));
+
+ public static Configuration getConfiguration() {
Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS,
3);
config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 12L);
- cluster = new LocalFlinkMiniCluster(config, false);
-
- cluster.start();
-
- env = new TestEnvironment(cluster, 6, false);
- }
-
- @AfterClass
- public static void shutdownCluster() {
- cluster.stop();
- cluster = null;
+ return config;
}
@Test
public void testFaultyAccumulator() throws Exception {
-
+ TestEnvironment env =
MINI_CLUSTER_RESOURCE.getTestEnvironment();
--- End diff --
You could also write `StreamExecutionEnvironment.getExecutionEnvironment()`.
---
