Github user sihuazhou commented on a diff in the pull request:
https://github.com/apache/flink/pull/5582#discussion_r192344507
--- Diff:
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
---
@@ -547,4 +549,30 @@ public boolean accept(File file, String s) {
return true;
}
}
+
+ private static class TestRocksDBStateBackend extends
RocksDBStateBackend {
+
+ public TestRocksDBStateBackend(AbstractStateBackend
checkpointStreamBackend, boolean enableIncrementalCheckpointing) {
+ super(checkpointStreamBackend,
enableIncrementalCheckpointing);
+ }
+
+ @Override
+ public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
+ Environment env,
+ JobID jobID,
+ String operatorIdentifier,
+ TypeSerializer<K> keySerializer,
+ int numberOfKeyGroups,
+ KeyGroupRange keyGroupRange,
+ TaskKvStateRegistry kvStateRegistry) throws IOException
{
+
+ AbstractKeyedStateBackend<K> keyedStateBackend =
super.createKeyedStateBackend(
+ env, jobID, operatorIdentifier, keySerializer,
numberOfKeyGroups, keyGroupRange, kvStateRegistry);
+
+ // We ignore the range deletions on production, but
when we are running the tests we shouldn't ignore it.
--- End diff --
Or maybe, the `chooseTheBestStateHandleToInit` could also return a non-null
instance if it's found the there are only a few invalid key-group that doesn't
belong to the target key-group, and comparing to iterate the valid records and
put them, it more cheap to just delete the invalid records, e.g. the target
key-group range is [0, 10] and there is one handle's key-group is [2, 11], this
case we may only need to delete(use the normal delete) the key-group 11 from
this handle.
---
