You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2021/03/31 10:19:14 UTC
[flink] branch master updated: [FLINK-21413][state] Clean
TtlMapState and TtlListState after all elements are expired
This is an automated email from the ASF dual-hosted git repository.
liyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1e17621 [FLINK-21413][state] Clean TtlMapState and TtlListState after all elements are expired
1e17621 is described below
commit 1e17621e65bfa8f4f3aff379118654ed77a82bd2
Author: liaojiayi <li...@bytedance.com>
AuthorDate: Thu Mar 18 18:55:43 2021 +0800
[FLINK-21413][state] Clean TtlMapState and TtlListState after all elements are expired
This closes #15016.
---
.../flink/runtime/state/ttl/TtlListState.java | 8 ++++-
.../flink/runtime/state/ttl/TtlMapState.java | 11 ++++++-
.../flink/runtime/state/ttl/TtlStateTestBase.java | 34 +++++++++++++++++++++-
.../runtime/state/ttl/TtlStateTestContextBase.java | 5 ++++
4 files changed, 55 insertions(+), 3 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlListState.java
index 5ab9813..7971275 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlListState.java
@@ -103,6 +103,11 @@ class TtlListState<K, N, T>
@Nullable
@Override
public List<TtlValue<T>> getUnexpiredOrNull(@Nonnull List<TtlValue<T>> ttlValues) {
+ // the update operation will clear the whole state if the list becomes empty after init
+ if (ttlValues.isEmpty()) {
+ return ttlValues;
+ }
+
long currentTimestamp = timeProvider.currentTimestamp();
List<TtlValue<T>> unexpired = new ArrayList<>(ttlValues.size());
TypeSerializer<TtlValue<T>> elementSerializer =
@@ -117,7 +122,8 @@ class TtlListState<K, N, T>
if (!unexpired.isEmpty()) {
return unexpired;
} else {
- return ttlValues.size() == unexpired.size() ? ttlValues : unexpired;
+ // list is not empty and all expired
+ return null;
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlMapState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlMapState.java
index d4fb7a8..bb1e20e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlMapState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlMapState.java
@@ -134,6 +134,10 @@ class TtlMapState<K, N, UK, UV>
@Nullable
@Override
public Map<UK, TtlValue<UV>> getUnexpiredOrNull(@Nonnull Map<UK, TtlValue<UV>> ttlValue) {
+ // the remove operation will clear the whole state if the map becomes empty after init
+ if (ttlValue.isEmpty()) {
+ return ttlValue;
+ }
Map<UK, TtlValue<UV>> unexpired = new HashMap<>();
TypeSerializer<TtlValue<UV>> valueSerializer =
((MapSerializer<UK, TtlValue<UV>>) original.getValueSerializer())
@@ -144,7 +148,12 @@ class TtlMapState<K, N, UK, UV>
unexpired.put(e.getKey(), valueSerializer.copy(e.getValue()));
}
}
- return ttlValue.size() == unexpired.size() ? ttlValue : unexpired;
+ if (!unexpired.isEmpty()) {
+ return unexpired;
+ } else {
+ // map is not empty but all values expired
+ return null;
+ }
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
index ef8fe2e..d6225ab 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
@@ -25,8 +25,10 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.heap.AbstractHeapState;
import org.apache.flink.runtime.state.heap.CopyOnWriteStateMap;
import org.apache.flink.runtime.state.internal.InternalKvState;
+import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StateMigrationException;
import org.junit.After;
@@ -151,9 +153,11 @@ public abstract class TtlStateTestBase {
private <S extends State> StateDescriptor<S, Object> createState() throws Exception {
StateDescriptor<S, Object> stateDescriptor = ctx().createStateDescriptor();
stateDescriptor.enableTimeToLive(ttlConfig);
+ String defaultNamespace = "defaultNamespace";
ctx().ttlState =
(InternalKvState<?, String, ?>)
- sbetc.createState(stateDescriptor, "defaultNamespace");
+ sbetc.createState(stateDescriptor, defaultNamespace);
+ ctx().setCurrentNamespace(defaultNamespace);
return stateDescriptor;
}
@@ -472,6 +476,26 @@ public abstract class TtlStateTestBase {
}
@Test
+ public void testIncrementalCleanupWholeState() throws Exception {
+ assumeTrue(incrementalCleanupSupported());
+ initTest(getConfBuilder(TTL).cleanupIncrementally(5, true).build());
+ timeProvider.time = 0;
+ // create enough keys to trigger incremental rehash
+ updateKeys(0, INC_CLEANUP_ALL_KEYS, ctx().updateEmpty);
+ // expire all state
+ timeProvider.time = 120;
+ // trigger state clean up
+ for (int i = 0; i < INC_CLEANUP_ALL_KEYS; i++) {
+ sbetc.setCurrentKey(Integer.toString(i));
+ }
+ // check all state cleaned up
+ for (int i = 0; i < INC_CLEANUP_ALL_KEYS; i++) {
+ sbetc.setCurrentKey(Integer.toString(i));
+ assertTrue("Original state should be cleared", isOriginalCleared());
+ }
+ }
+
+ @Test
public void testIncrementalCleanup() throws Exception {
assumeTrue(incrementalCleanupSupported());
@@ -558,6 +582,14 @@ public abstract class TtlStateTestBase {
}
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private boolean isOriginalCleared() {
+ InternalKvState<?, String, ?> original =
+ ((AbstractTtlDecorator<InternalKvState<?, String, ?>>) ctx.ttlState).original;
+ Preconditions.checkState(original instanceof AbstractHeapState);
+ return ((AbstractHeapState) original).getStateTable().get(ctx.currentNamespace) == null;
+ }
+
@After
public void tearDown() throws Exception {
sbetc.dispose();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestContextBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestContextBase.java
index a98cd56..448c5d5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestContextBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestContextBase.java
@@ -37,6 +37,7 @@ public abstract class TtlStateTestContextBase<S extends InternalKvState<?, Strin
GV getUpdateExpired;
public GV emptyValue = null;
+ public String currentNamespace = "defaultNamespace";
abstract void initTestValues();
@@ -60,4 +61,8 @@ public abstract class TtlStateTestContextBase<S extends InternalKvState<?, Strin
public String toString() {
return this.getClass().getSimpleName();
}
+
+ public void setCurrentNamespace(String currentNamespace) {
+ this.currentNamespace = currentNamespace;
+ }
}