diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java index effbbc52fc81d..f649b8f7df378 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java @@ -38,12 +38,10 @@ import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; -import org.apache.flink.testutils.junit.FailsWithAdaptiveScheduler; import org.apache.flink.util.Collector; import org.apache.commons.lang3.ArrayUtils; import org.junit.Test; -import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; @@ -105,7 +103,6 @@ * */ @RunWith(Parameterized.class) -@Category(FailsWithAdaptiveScheduler.class) // FLINK-21689 public class UnalignedCheckpointITCase extends UnalignedCheckpointTestBase { enum Topology implements DagCreator { PIPELINE { diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java index c6201faff07cd..25eef0480b1c1 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java @@ -104,8 +104,16 @@ import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT; import static org.apache.flink.util.Preconditions.checkState; -/** Base class for tests related to unaligned checkpoints. */ -@Category(FailsWithAdaptiveScheduler.class) // FLINK-21689 +/** + * Base class for tests related to unaligned checkpoints. + * + *

This test base relies on restarting the subtasks within the scheduler to trigger a reset of + * the operators. The operator reset is counted in the LongSource. The job will terminate if the + * number of expected restarts is reached. The AdaptiveScheduler won't trigger the operator reset + * resulting in the test running forever. This is why this test suite is disabled for the {@link + * org.apache.flink.runtime.scheduler.adaptive.AdaptiveScheduler}. + */ +@Category(FailsWithAdaptiveScheduler.class) public abstract class UnalignedCheckpointTestBase extends TestLogger { protected static final Logger LOG = LoggerFactory.getLogger(UnalignedCheckpointTestBase.class); protected static final String NUM_INPUTS = "inputs_";