Skip to content

Commit 0d0c185

Browse files
frediafredia
authored
Revert "[FLINK-37158][tests] Introduce ForSt to existing ITCases (#26000)" (#26291)
This reverts commit 6099697.
1 parent 6691348 commit 0d0c185

File tree

5 files changed

+8
-169
lines changed

5 files changed

+8
-169
lines changed

‎flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackend.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -608,8 +608,7 @@ public RunnableFuture<SnapshotResult<KeyedStateHandle>> snapshot(
608608
@Nonnull
609609
@Override
610610
public SavepointResources<K> savepoint() throws Exception {
611-
throw new UnsupportedOperationException(
612-
"Canonical savepoints are not supported by ForSt State Backend.");
611+
throw new UnsupportedOperationException("This method is not supported.");
613612
}
614613

615614
@Override

‎flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java

+1-3
Original file line numberDiff line numberDiff line change
@@ -116,9 +116,7 @@ public static Collection<Object[]> data() {
116116
new Object[][] {
117117
{"rocksdb", false},
118118
{"rocksdb", true},
119-
{"hashmap", false},
120-
{"forst", false},
121-
{"forst", true}
119+
{"hashmap", false}
122120
});
123121
}
124122

‎flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java

-34
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,6 @@
3535
import org.apache.flink.core.fs.Path;
3636
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
3737
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
38-
import org.apache.flink.state.forst.ForStOptions;
39-
import org.apache.flink.state.forst.ForStStateBackend;
4038
import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend;
4139
import org.apache.flink.state.rocksdb.RocksDBOptions;
4240
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -112,7 +110,6 @@ enum StateBackendEnum {
112110
ROCKSDB_FULL,
113111
ROCKSDB_INCREMENTAL,
114112
ROCKSDB_INCREMENTAL_ZK,
115-
FORST_INCREMENTAL
116113
}
117114

118115
@Parameterized.Parameters(name = "statebackend type ={0}")
@@ -194,14 +191,6 @@ private Configuration getConfiguration() throws Exception {
194191
setupRocksDB(config, 16, true);
195192
break;
196193
}
197-
case FORST_INCREMENTAL:
198-
{
199-
config.set(
200-
ForStOptions.TIMER_SERVICE_FACTORY,
201-
ForStStateBackend.PriorityQueueStateType.ForStDB);
202-
setupForSt(config, 16);
203-
break;
204-
}
205194
default:
206195
throw new IllegalStateException("No backend selected.");
207196
}
@@ -240,29 +229,6 @@ private void setupRocksDB(
240229
config.set(RocksDBOptions.LOCAL_DIRECTORIES, rocksDb);
241230
}
242231

243-
private void setupForSt(Configuration config, int fileSizeThreshold) throws IOException {
244-
// Configure the managed memory size as 64MB per slot for rocksDB state backend.
245-
config.set(
246-
TaskManagerOptions.MANAGED_MEMORY_SIZE,
247-
MemorySize.ofMebiBytes(PARALLELISM / NUM_OF_TASK_MANAGERS * 64));
248-
249-
final String forstdb = tempFolder.newFolder().getAbsolutePath();
250-
final File backups = tempFolder.newFolder().getAbsoluteFile();
251-
// we use the fs backend with small threshold here to test the behaviour with file
252-
// references, not self contained byte handles
253-
config.set(StateBackendOptions.STATE_BACKEND, "forst");
254-
config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
255-
config.set(
256-
CheckpointingOptions.CHECKPOINTS_DIRECTORY,
257-
Path.fromLocalFile(backups).toUri().toString());
258-
if (fileSizeThreshold != -1) {
259-
config.set(
260-
CheckpointingOptions.FS_SMALL_FILE_THRESHOLD,
261-
MemorySize.parse(fileSizeThreshold + "b"));
262-
}
263-
config.set(ForStOptions.LOCAL_DIRECTORIES, forstdb);
264-
}
265-
266232
protected Configuration createClusterConfig() throws IOException {
267233
TemporaryFolder temporaryFolder = new TemporaryFolder();
268234
temporaryFolder.create();

‎flink-tests/src/test/java/org/apache/flink/test/checkpointing/KeyedStateCheckpointingITCase.java

-18
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,10 @@
2525
import org.apache.flink.api.common.state.ValueStateDescriptor;
2626
import org.apache.flink.api.java.functions.KeySelector;
2727
import org.apache.flink.api.java.tuple.Tuple2;
28-
import org.apache.flink.configuration.CheckpointingOptions;
2928
import org.apache.flink.configuration.Configuration;
3029
import org.apache.flink.configuration.MemorySize;
31-
import org.apache.flink.configuration.StateBackendOptions;
3230
import org.apache.flink.configuration.TaskManagerOptions;
3331
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
34-
import org.apache.flink.state.forst.ForStOptions;
3532
import org.apache.flink.state.rocksdb.RocksDBOptions;
3633
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
3734
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -146,21 +143,6 @@ public void testWithRocksDbBackendIncremental() throws Exception {
146143
testProgramWithBackend(env);
147144
}
148145

149-
@Test
150-
public void testWithForStBackendIncremental() throws Exception {
151-
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
152-
env.configure(
153-
new Configuration()
154-
.set(StateBackendOptions.STATE_BACKEND, "forst")
155-
.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true)
156-
.set(
157-
ForStOptions.LOCAL_DIRECTORIES,
158-
tmpFolder.newFolder().getAbsolutePath()));
159-
CheckpointStorageUtils.configureFileSystemCheckpointStorage(
160-
env, tmpFolder.newFolder().toURI().toString());
161-
testProgramWithBackend(env);
162-
}
163-
164146
// ------------------------------------------------------------------------
165147

166148
protected void testProgramWithBackend(StreamExecutionEnvironment env) throws Exception {

‎flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescaleCheckpointManuallyITCase.java

+6-112
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,9 @@
1919
package org.apache.flink.test.checkpointing;
2020

2121
import org.apache.flink.api.common.JobID;
22-
import org.apache.flink.api.common.functions.OpenContext;
2322
import org.apache.flink.api.common.functions.RichFlatMapFunction;
2423
import org.apache.flink.api.common.state.ValueState;
2524
import org.apache.flink.api.common.state.ValueStateDescriptor;
26-
import org.apache.flink.api.common.state.v2.StateFuture;
27-
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
2825
import org.apache.flink.api.connector.sink2.Sink;
2926
import org.apache.flink.api.connector.sink2.SinkWriter;
3027
import org.apache.flink.api.connector.sink2.WriterInitContext;
@@ -43,10 +40,8 @@
4340
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
4441
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
4542
import org.apache.flink.streaming.api.datastream.DataStream;
46-
import org.apache.flink.streaming.api.datastream.KeyedStream;
4743
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4844
import org.apache.flink.streaming.api.functions.source.legacy.RichParallelSourceFunction;
49-
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
5045
import org.apache.flink.test.util.MiniClusterWithClientResource;
5146
import org.apache.flink.testutils.junit.SharedObjects;
5247
import org.apache.flink.testutils.junit.SharedReference;
@@ -61,12 +56,8 @@
6156
import org.junit.Rule;
6257
import org.junit.Test;
6358
import org.junit.rules.TemporaryFolder;
64-
import org.junit.runner.RunWith;
65-
import org.junit.runners.Parameterized;
6659

6760
import java.io.IOException;
68-
import java.util.Arrays;
69-
import java.util.Collection;
7061
import java.util.Collections;
7162
import java.util.HashSet;
7263
import java.util.Optional;
@@ -83,7 +74,6 @@
8374
* NotifyingDefiniteKeySource, SubtaskIndexFlatMapper and CollectionSink refer to RescalingITCase,
8475
* because the static fields in these classes can not be shared.
8576
*/
86-
@RunWith(Parameterized.class)
8777
public class RescaleCheckpointManuallyITCase extends TestLogger {
8878

8979
private static final int NUM_TASK_MANAGERS = 2;
@@ -94,24 +84,10 @@ public class RescaleCheckpointManuallyITCase extends TestLogger {
9484

9585
@ClassRule public static TemporaryFolder temporaryFolder = new TemporaryFolder();
9686

97-
@Parameterized.Parameter(0)
98-
public String statebackendType;
99-
100-
@Parameterized.Parameter(1)
101-
public boolean enableAsyncState;
102-
103-
@Parameterized.Parameters(name = "statebackend type ={0}, enableAsyncState={1}")
104-
public static Collection<Object[]> parameter() {
105-
return Arrays.asList(
106-
new Object[][] {
107-
{"forst", true}, {"forst", false}, {"rocksdb", true}, {"rocksdb", false}
108-
});
109-
}
110-
11187
@Before
11288
public void setup() throws Exception {
11389
Configuration config = new Configuration();
114-
config.set(StateBackendOptions.STATE_BACKEND, statebackendType);
90+
config.set(StateBackendOptions.STATE_BACKEND, "rocksdb");
11591
config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
11692

11793
cluster =
@@ -287,7 +263,7 @@ private JobGraph createJobGraphWithKeyedState(
287263

288264
SharedReference<JobID> jobID = sharedObjects.add(new JobID());
289265
SharedReference<MiniCluster> miniClusterRef = sharedObjects.add(miniCluster);
290-
KeyedStream<Integer, Integer> input =
266+
DataStream<Integer> input =
291267
env.addSource(
292268
new NotifyingDefiniteKeySource(
293269
numberKeys, numberElements, failAfterEmission) {
@@ -324,18 +300,10 @@ public Integer getKey(Integer value) {
324300
return value;
325301
}
326302
});
327-
if (enableAsyncState) {
328-
input.enableAsyncState();
329-
DataStream<Tuple2<Integer, Integer>> result =
330-
input.flatMap(new AsyncSubtaskIndexFlatMapper(numberElementsExpect));
303+
DataStream<Tuple2<Integer, Integer>> result =
304+
input.flatMap(new SubtaskIndexFlatMapper(numberElementsExpect));
331305

332-
result.sinkTo(new CollectionSink<>());
333-
} else {
334-
DataStream<Tuple2<Integer, Integer>> result =
335-
input.flatMap(new SubtaskIndexFlatMapper(numberElementsExpect));
336-
337-
result.sinkTo(new CollectionSink<>());
338-
}
306+
result.sinkTo(new CollectionSink<>());
339307

340308
return env.getStreamGraph().getJobGraph(env.getClass().getClassLoader(), jobID.get());
341309
}
@@ -381,9 +349,8 @@ public void run(SourceContext<Integer> ctx) throws Exception {
381349
} else {
382350
boolean newCheckpoint = false;
383351
long waited = 0L;
384-
running = false;
385352
// maximum wait 5min
386-
while (!newCheckpoint && waited < 300000L) {
353+
while (!newCheckpoint && waited < 30000L) {
387354
synchronized (ctx.getCheckpointLock()) {
388355
newCheckpoint = waitCheckpointCompleted();
389356
}
@@ -456,79 +423,6 @@ public void initializeState(FunctionInitializationContext context) throws Except
456423
}
457424
}
458425

459-
private static class AsyncSubtaskIndexFlatMapper
460-
extends RichFlatMapFunction<Integer, Tuple2<Integer, Integer>>
461-
implements CheckpointedFunction {
462-
463-
private static final long serialVersionUID = 1L;
464-
465-
private transient org.apache.flink.api.common.state.v2.ValueState<Integer> counter;
466-
private transient org.apache.flink.api.common.state.v2.ValueState<Integer> sum;
467-
468-
private final int numberElements;
469-
470-
public AsyncSubtaskIndexFlatMapper(int numberElements) {
471-
this.numberElements = numberElements;
472-
}
473-
474-
@Override
475-
public void flatMap(Integer value, Collector<Tuple2<Integer, Integer>> out)
476-
throws Exception {
477-
StateFuture<Integer> counterFuture =
478-
counter.asyncValue()
479-
.thenCompose(
480-
(Integer c) -> {
481-
int updated = c == null ? 1 : c + 1;
482-
return counter.asyncUpdate(updated)
483-
.thenApply(nothing -> updated);
484-
});
485-
StateFuture<Integer> sumFuture =
486-
sum.asyncValue()
487-
.thenCompose(
488-
(Integer s) -> {
489-
int updated = s == null ? value : s + value;
490-
return sum.asyncUpdate(updated)
491-
.thenApply(nothing -> updated);
492-
});
493-
494-
counterFuture.thenCombine(
495-
sumFuture,
496-
(c, s) -> {
497-
if (c == numberElements) {
498-
out.collect(
499-
Tuple2.of(
500-
getRuntimeContext()
501-
.getTaskInfo()
502-
.getIndexOfThisSubtask(),
503-
s));
504-
}
505-
return null;
506-
});
507-
}
508-
509-
@Override
510-
public void snapshotState(FunctionSnapshotContext context) throws Exception {
511-
// all managed, nothing to do.
512-
}
513-
514-
@Override
515-
public void initializeState(FunctionInitializationContext context) throws Exception {}
516-
517-
@Override
518-
public void open(OpenContext openContext) throws Exception {
519-
counter =
520-
((StreamingRuntimeContext) getRuntimeContext())
521-
.getValueState(
522-
new org.apache.flink.api.common.state.v2.ValueStateDescriptor<>(
523-
"counter", BasicTypeInfo.INT_TYPE_INFO));
524-
sum =
525-
((StreamingRuntimeContext) getRuntimeContext())
526-
.getValueState(
527-
new org.apache.flink.api.common.state.v2.ValueStateDescriptor<>(
528-
"sum", BasicTypeInfo.INT_TYPE_INFO));
529-
}
530-
}
531-
532426
private static class CollectionSink<IN> implements Sink<IN> {
533427

534428
private static final ConcurrentHashMap<JobID, CollectionSinkWriter<?>> writers =

0 commit comments

Comments
 (0)