dawidwys commented on a change in pull request #18729:
URL: https://github.com/apache/flink/pull/18729#discussion_r805687984
##########
File path:
flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointFormatITCase.java
##########
@@ -78,138 +82,151 @@
LoggerAuditingExtension loggerAuditingExtension =
new LoggerAuditingExtension(SavepointFormatITCase.class,
Level.INFO);
- private static Stream<Arguments> parameters() {
- return Stream.of(
- Arguments.of(
- SavepointFormatType.CANONICAL,
- HEAP,
- (Consumer<KeyedStateHandle>)
- keyedState ->
- assertThat(
- keyedState,
-
instanceOf(SavepointKeyedStateHandle.class))),
- Arguments.of(
- SavepointFormatType.NATIVE,
- HEAP,
- (Consumer<KeyedStateHandle>)
- keyedState ->
- assertThat(
- keyedState,
-
instanceOf(KeyGroupsStateHandle.class))),
- Arguments.of(
- SavepointFormatType.CANONICAL,
- ROCKSDB_FULL_SNAPSHOTS,
- (Consumer<KeyedStateHandle>)
- keyedState ->
- assertThat(
- keyedState,
-
instanceOf(SavepointKeyedStateHandle.class))),
- Arguments.of(
- SavepointFormatType.NATIVE,
- ROCKSDB_FULL_SNAPSHOTS,
- (Consumer<KeyedStateHandle>)
- keyedState ->
- assertThat(
- keyedState,
-
instanceOf(KeyGroupsStateHandle.class))),
- Arguments.of(
- SavepointFormatType.CANONICAL,
- ROCKSDB_INCREMENTAL_SNAPSHOTS,
- (Consumer<KeyedStateHandle>)
- keyedState ->
- assertThat(
- keyedState,
-
instanceOf(SavepointKeyedStateHandle.class))),
- Arguments.of(
- SavepointFormatType.NATIVE,
- ROCKSDB_INCREMENTAL_SNAPSHOTS,
- (Consumer<KeyedStateHandle>)
- keyedState ->
- assertThat(
- keyedState,
- instanceOf(
-
IncrementalRemoteKeyedStateHandle.class))));
+ private static List<Arguments> parameters() {
+ // iterate through all combinations of backends, isIncremental,
isChangelogEnabled
+ List<Arguments> result = new LinkedList<>();
+ for (BiFunction<Boolean, Boolean, StateBackendConfig> builder :
+ StateBackendConfig.builders) {
+ for (boolean incremental : new boolean[] {true, false}) {
+ for (boolean changelog : new boolean[] {true, false}) {
+ for (SavepointFormatType formatType :
SavepointFormatType.values()) {
+ result.add(Arguments.of(formatType,
builder.apply(incremental, changelog)));
+ }
+ }
+ }
+ }
+ return result;
+ }
+
+ private void validateState(
+ KeyedStateHandle state,
+ SavepointFormatType formatType,
+ StateBackendConfig backendConfig) {
+ if (formatType == SavepointFormatType.CANONICAL) {
+ assertThat(state, instanceOf(SavepointKeyedStateHandle.class));
+ } else if (backendConfig.isChangelogEnabled()) {
+ assertThat(state, instanceOf(ChangelogStateBackendHandle.class));
+ for (KeyedStateHandle nestedState :
+ ((ChangelogStateBackendHandle)
state).getMaterializedStateHandles()) {
+ validateNativeNonChangelogState(nestedState, backendConfig);
+ }
+ } else {
+ validateNativeNonChangelogState(state, backendConfig);
+ }
+ }
+
+ private void validateNativeNonChangelogState(
+ KeyedStateHandle state, StateBackendConfig backendConfig) {
+ if (backendConfig.isIncremental()) {
+ assertThat(state,
instanceOf(IncrementalRemoteKeyedStateHandle.class));
+ } else {
+ assertThat(state, instanceOf(KeyGroupsStateHandle.class));
+ }
}
private abstract static class StateBackendConfig {
+ protected final boolean changelogEnabled;
+ protected final boolean incremental;
+
+ protected StateBackendConfig(boolean changelogEnabled, boolean
incremental) {
+ this.changelogEnabled = changelogEnabled;
+ this.incremental = incremental;
+ }
+
public abstract String getName();
- public abstract Configuration getConfiguration();
+ public Configuration getConfiguration() {
+ Configuration stateBackendConfig = new Configuration();
+ stateBackendConfig.setString(StateBackendOptions.STATE_BACKEND,
getConfigName());
+
stateBackendConfig.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS,
incremental);
+
stateBackendConfig.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG,
changelogEnabled);
+ return stateBackendConfig;
+ }
public int getCheckpointsBeforeSavepoint() {
return 0;
}
+ protected abstract String getConfigName();
+
@Override
public final String toString() {
- return getName();
+ return String.format(
+ "%s, incremental: %b, changelog: %b", getName(),
incremental, changelogEnabled);
}
- }
- private static final StateBackendConfig HEAP =
- new StateBackendConfig() {
- @Override
- public String getName() {
- return "HEAP";
- }
-
- @Override
- public Configuration getConfiguration() {
- Configuration stateBackendConfig = new Configuration();
-
stateBackendConfig.setString(StateBackendOptions.STATE_BACKEND, "filesystem");
- stateBackendConfig.set(
- CheckpointingOptions.FS_SMALL_FILE_THRESHOLD,
MemorySize.ZERO);
- return stateBackendConfig;
- }
- };
-
- private static final StateBackendConfig ROCKSDB_FULL_SNAPSHOTS =
- new StateBackendConfig() {
- @Override
- public String getName() {
- return "ROCKSDB_FULL_SNAPSHOTS";
- }
+ private static final List<BiFunction<Boolean, Boolean,
StateBackendConfig>> builders =
+ asList(SavepointFormatITCase::getRocksdb,
SavepointFormatITCase::heap);
- @Override
- public Configuration getConfiguration() {
- Configuration stateBackendConfig = new Configuration();
-
stateBackendConfig.setString(StateBackendOptions.STATE_BACKEND, "rocksdb");
- stateBackendConfig.set(
- CheckpointingOptions.FS_SMALL_FILE_THRESHOLD,
MemorySize.ZERO);
-
stateBackendConfig.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, false);
- return stateBackendConfig;
- }
- };
+ public abstract boolean isIncremental();
- private static final StateBackendConfig ROCKSDB_INCREMENTAL_SNAPSHOTS =
- new StateBackendConfig() {
- @Override
- public String getName() {
- return "ROCKSDB_INCREMENTAL_SNAPSHOTS";
- }
+ private boolean isChangelogEnabled() {
+ return changelogEnabled;
+ }
+ }
- @Override
- public int getCheckpointsBeforeSavepoint() {
- return 1;
- }
+ private static StateBackendConfig heap(boolean incremental, boolean
changelogEnabled) {
+ return new StateBackendConfig(changelogEnabled, incremental /* ignored
for now */) {
Review comment:
I'll leave it up to you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
