liming30 commented on code in PR #20405:
URL: https://github.com/apache/flink/pull/20405#discussion_r943331650
##########
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java:
##########
@@ -637,11 +644,73 @@ public void testMapStateClear() throws Exception {
throw new RocksDBException("Artificial failure");
})
.when(keyedStateBackend.db)
- .newIterator(any(ColumnFamilyHandle.class),
any(ReadOptions.class));
+ .deleteRange(any(ColumnFamilyHandle.class), any(byte[].class),
any(byte[].class));
state.clear();
}
+ @Test
+ public void testMapStateClearCorrectly() throws Exception {
+ verifyMapStateClear(Byte.MAX_VALUE + 1); // one byte prefix
+
verifyMapStateClear(KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM); //
two byte prefix
+ }
+
+ public void verifyMapStateClear(int maxParallelism) throws Exception {
+ try {
+ this.maxParallelism = maxParallelism;
+ setupRocksKeyedStateBackend();
+ MapStateDescriptor<Integer, String> kvId =
+ new MapStateDescriptor<>("id", Integer.class,
String.class);
+ MapState<Integer, String> state =
+ keyedStateBackend.getPartitionedState(
+ VoidNamespace.INSTANCE,
VoidNamespaceSerializer.INSTANCE, kvId);
+ keyedStateBackend = spy(keyedStateBackend);
+ doAnswer(
+ invocationOnMock -> { // ensure that each KeyGroup
can use the same Key
+ keyedStateBackend
+ .getKeyContext()
+
.setCurrentKey(invocationOnMock.getArgument(0));
+ keyedStateBackend
+ .getSharedRocksKeyBuilder()
+ .setKeyAndKeyGroup(
+
keyedStateBackend.getCurrentKey(),
+
keyedStateBackend.getCurrentKeyGroupIndex());
+ return null;
+ })
+ .when(keyedStateBackend)
+ .setCurrentKey(any());
+
+ Set<Integer> testKeyGroups =
+ new HashSet<>(Arrays.asList(0, maxParallelism / 2,
maxParallelism - 1));
+ for (int testKeyGroup : testKeyGroups) { // initialize
+ if (testKeyGroup >= maxParallelism - 1) {
+ break;
+ }
+ keyedStateBackend.setCurrentKeyGroupIndex(testKeyGroup);
+ keyedStateBackend.setCurrentKey(testKeyGroup);
+ state.put(testKeyGroup, "retain " + testKeyGroup);
+ keyedStateBackend.setCurrentKey(-1); // 0xffff for the key
+ state.put(testKeyGroup, "clear " + testKeyGroup);
+ }
+
+ for (int testKeyGroup : testKeyGroups) { // test for clear
+ if (testKeyGroup >= maxParallelism - 1) {
+ break;
+ }
+ keyedStateBackend.setCurrentKeyGroupIndex(testKeyGroup);
+ keyedStateBackend.setCurrentKey(-1);
+ assertEquals("clear " + testKeyGroup, state.get(testKeyGroup));
+ state.clear();
+ assertNull(state.get(testKeyGroup));
+ keyedStateBackend.setCurrentKey(testKeyGroup);
+ assertEquals("retain " + testKeyGroup,
state.get(testKeyGroup));
+ }
Review Comment:
Resolved, please help to review it when you have time.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
