You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2019/06/05 18:40:46 UTC
[flink] branch release-1.8 updated: [FLINK-12476] [State TTL]
Consider setting a default background cleanup strategy in StateTtlConfig
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch release-1.8
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.8 by this push:
new 877411a [FLINK-12476] [State TTL] Consider setting a default background cleanup strategy in StateTtlConfig
877411a is described below
commit 877411a56ad921935851e6eda3fcee0535597d73
Author: yanghua <ya...@gmail.com>
AuthorDate: Thu May 16 11:35:13 2019 +0800
[FLINK-12476] [State TTL] Consider setting a default background cleanup strategy in StateTtlConfig
At the moment we have two efficient background cleanup strategies: incremental for heap and compaction filter for RocksDB. StateTtlConfig has 2 methods to activate them: cleanupIncrementally and cleanupInRocksdbCompactFilter. Each is activated only for certain backend type and inactive for other. They have different tuning parameters.
The idea is to add method cleanupInBackground which would activate default background cleanup. User does not need to think then about details or used backend if not needed. Depending on actually used backend, the corresponding cleanup will kick in. If original strategy is not set with cleanupIncrementally and cleanupInRocksdbCompactFilter then backends should check whether default background cleanup is activated and if so, use it with default parameters.
We can also deprecate the parameterless cleanupInRocksdbCompactFilter() in favour of this new method.
---
docs/dev/stream/state/state.md | 39 ++++++++++++--
.../flink/api/common/state/StateTtlConfig.java | 63 ++++++++++++++++------
.../flink/api/common/state/StateTtlConfigTest.java | 58 ++++++++++++++++++++
.../tests/DataStreamStateTTLTestProgram.java | 3 +-
.../state/ttl/RocksDBTtlStateTestBase.java | 2 +-
5 files changed, 141 insertions(+), 24 deletions(-)
diff --git a/docs/dev/stream/state/state.md b/docs/dev/stream/state/state.md
index be748d4..e99a7b8 100644
--- a/docs/dev/stream/state/state.md
+++ b/docs/dev/stream/state/state.md
@@ -400,6 +400,35 @@ This option is not applicable for the incremental checkpointing in the RocksDB s
- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
e.g. after restart from savepoint.
+#### Cleanup in background
+
+Besides cleanup in full snapshot, you can also activate the cleanup in background. The following option
+will activate a default background cleanup in StateTtlConfig if it is supported for the used backend:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+import org.apache.flink.api.common.state.StateTtlConfig;
+StateTtlConfig ttlConfig = StateTtlConfig
+ .newBuilder(Time.seconds(1))
+ .cleanupInBackground()
+ .build();
+{% endhighlight %}
+</div>
+ <div data-lang="scala" markdown="1">
+{% highlight scala %}
+import org.apache.flink.api.common.state.StateTtlConfig
+val ttlConfig = StateTtlConfig
+ .newBuilder(Time.seconds(1))
+ .cleanupInBackground
+ .build
+{% endhighlight %}
+</div>
+</div>
+
+For more fine-grained control over some special cleanup in background, you can configure it separately as described below.
+Currently, heap state backend relies on incremental cleanup and RocksDB backend uses compaction filter for background cleanup.
+
##### Incremental cleanup
Another option is to trigger cleanup of some state entries incrementally.
@@ -435,6 +464,7 @@ val ttlConfig = StateTtlCon fig
This strategy has two parameters. The first one is number of checked state entries per each cleanup triggering.
If enabled, it is always triggered per each state access.
The second parameter defines whether to trigger cleanup additionally per each record processing.
+If you enable the default background cleanup then this strategy will be activated for heap backend with 5 checked entries and without cleanup per record processing.
**Notes:**
- If no access happens to the state or no records are processed, expired state will persist.
@@ -465,7 +495,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter()
+ .cleanupInRocksdbCompactFilter(1000)
.build();
{% endhighlight %}
</div>
@@ -476,18 +506,19 @@ import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter
+ .cleanupInRocksdbCompactFilter(1000)
.build
{% endhighlight %}
</div>
</div>
RocksDB compaction filter will query current timestamp, used to check expiration, from Flink every time
-after processing certain number of state entries. This number is 1000 by default.
-You can optionally change it and pass a custom value to
+after processing certain number of state entries.
+You can change it and pass a custom value to
`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` method.
Updating the timestamp more often can improve cleanup speed
but it decreases compaction performance because it uses JNI call from native code.
+If you enable the default background cleanup then this strategy will be activated for RocksDB backend and the current timestamp will be queried each time 1000 entries have been processed.
You can activate debug logs from the native code of RocksDB filter
by activating debug level for `FlinkCompactionFilter`:
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/StateTtlConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/state/StateTtlConfig.java
index 5bb44d1..83ee879 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/state/StateTtlConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/state/StateTtlConfig.java
@@ -28,6 +28,9 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.EnumMap;
+import static org.apache.flink.api.common.state.StateTtlConfig.CleanupStrategies.EMPTY_STRATEGY;
+import static org.apache.flink.api.common.state.StateTtlConfig.IncrementalCleanupStrategy.DEFAULT_INCREMENTAL_CLEANUP_STRATEGY;
+import static org.apache.flink.api.common.state.StateTtlConfig.RocksdbCompactFilterCleanupStrategy.DEFAULT_ROCKSDB_COMPACT_FILTER_CLEANUP_STRATEGY;
import static org.apache.flink.api.common.state.StateTtlConfig.StateVisibility.NeverReturnExpired;
import static org.apache.flink.api.common.state.StateTtlConfig.TimeCharacteristic.ProcessingTime;
import static org.apache.flink.api.common.state.StateTtlConfig.UpdateType.OnCreateAndWrite;
@@ -152,7 +155,9 @@ public class StateTtlConfig implements Serializable {
private StateVisibility stateVisibility = NeverReturnExpired;
private TimeCharacteristic timeCharacteristic = ProcessingTime;
private Time ttl;
- private CleanupStrategies cleanupStrategies = new CleanupStrategies();
+ private boolean isCleanupInBackground = false;
+ private final EnumMap<CleanupStrategies.Strategies, CleanupStrategies.CleanupStrategy> strategies =
+ new EnumMap<>(CleanupStrategies.Strategies.class);
public Builder(@Nonnull Time ttl) {
this.ttl = ttl;
@@ -219,7 +224,7 @@ public class StateTtlConfig implements Serializable {
/** Cleanup expired state in full snapshot on checkpoint. */
@Nonnull
public Builder cleanupFullSnapshot() {
- cleanupStrategies.activate(CleanupStrategies.Strategies.FULL_STATE_SCAN_SNAPSHOT);
+ strategies.put(CleanupStrategies.Strategies.FULL_STATE_SCAN_SNAPSHOT, EMPTY_STRATEGY);
return this;
}
@@ -254,7 +259,7 @@ public class StateTtlConfig implements Serializable {
public Builder cleanupIncrementally(
@Nonnegative int cleanupSize,
boolean runCleanupForEveryRecord) {
- cleanupStrategies.activate(
+ strategies.put(
CleanupStrategies.Strategies.INCREMENTAL_CLEANUP,
new IncrementalCleanupStrategy(cleanupSize, runCleanupForEveryRecord));
return this;
@@ -274,8 +279,10 @@ public class StateTtlConfig implements Serializable {
* https://github.com/facebook/rocksdb/blob/master/include/rocksdb/compaction_filter.h#L69
* It means that the TTL filter should be tested for List state taking into account this caveat.
*
+ * @deprecated Use more general configuration method {@link #cleanupInBackground()} instead
*/
@Nonnull
+ @Deprecated
public Builder cleanupInRocksdbCompactFilter() {
return cleanupInRocksdbCompactFilter(1000L);
}
@@ -292,12 +299,24 @@ public class StateTtlConfig implements Serializable {
*/
@Nonnull
public Builder cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries) {
- cleanupStrategies.activate(CleanupStrategies.Strategies.ROCKSDB_COMPACTION_FILTER,
+ strategies.put(
+ CleanupStrategies.Strategies.ROCKSDB_COMPACTION_FILTER,
new RocksdbCompactFilterCleanupStrategy(queryTimeAfterNumEntries));
return this;
}
/**
+ * Enable cleanup of expired state in background.
+ *
+ * <p>Depending on actually used backend, the corresponding cleanup will kick in if supported.
+ */
+ @Nonnull
+ public Builder cleanupInBackground() {
+ isCleanupInBackground = true;
+ return this;
+ }
+
+ /**
* Sets the ttl time.
* @param ttl The ttl time.
*/
@@ -314,7 +333,7 @@ public class StateTtlConfig implements Serializable {
stateVisibility,
timeCharacteristic,
ttl,
- cleanupStrategies);
+ new CleanupStrategies(strategies, isCleanupInBackground));
}
}
@@ -328,7 +347,11 @@ public class StateTtlConfig implements Serializable {
public static class CleanupStrategies implements Serializable {
private static final long serialVersionUID = -1617740467277313524L;
- private static final CleanupStrategy EMPTY_STRATEGY = new EmptyCleanupStrategy();
+ static final CleanupStrategy EMPTY_STRATEGY = new EmptyCleanupStrategy();
+
+ private final boolean isCleanupInBackground;
+
+ private final EnumMap<Strategies, CleanupStrategy> strategies;
/** Fixed strategies ordinals in {@code strategies} config field. */
enum Strategies {
@@ -346,32 +369,33 @@ public class StateTtlConfig implements Serializable {
private static final long serialVersionUID = 1373998465131443873L;
}
- final EnumMap<Strategies, CleanupStrategy> strategies = new EnumMap<>(Strategies.class);
-
- public void activate(Strategies strategy) {
- activate(strategy, EMPTY_STRATEGY);
- }
-
- public void activate(Strategies strategy, CleanupStrategy config) {
- strategies.put(strategy, config);
+ private CleanupStrategies(EnumMap<Strategies, CleanupStrategy> strategies, boolean isCleanupInBackground) {
+ this.strategies = strategies;
+ this.isCleanupInBackground = isCleanupInBackground;
}
public boolean inFullSnapshot() {
return strategies.containsKey(Strategies.FULL_STATE_SCAN_SNAPSHOT);
}
+ public boolean isCleanupInBackground() {
+ return isCleanupInBackground;
+ }
+
@Nullable
public IncrementalCleanupStrategy getIncrementalCleanupStrategy() {
- return (IncrementalCleanupStrategy) strategies.get(Strategies.INCREMENTAL_CLEANUP);
+ IncrementalCleanupStrategy defaultStrategy = isCleanupInBackground ? DEFAULT_INCREMENTAL_CLEANUP_STRATEGY : null;
+ return (IncrementalCleanupStrategy) strategies.getOrDefault(Strategies.INCREMENTAL_CLEANUP, defaultStrategy);
}
public boolean inRocksdbCompactFilter() {
- return strategies.containsKey(Strategies.ROCKSDB_COMPACTION_FILTER);
+ return getRocksdbCompactFilterCleanupStrategy() != null;
}
@Nullable
public RocksdbCompactFilterCleanupStrategy getRocksdbCompactFilterCleanupStrategy() {
- return (RocksdbCompactFilterCleanupStrategy) strategies.get(Strategies.ROCKSDB_COMPACTION_FILTER);
+ RocksdbCompactFilterCleanupStrategy defaultStrategy = isCleanupInBackground ? DEFAULT_ROCKSDB_COMPACT_FILTER_CLEANUP_STRATEGY : null;
+ return (RocksdbCompactFilterCleanupStrategy) strategies.getOrDefault(Strategies.ROCKSDB_COMPACTION_FILTER, defaultStrategy);
}
}
@@ -379,6 +403,8 @@ public class StateTtlConfig implements Serializable {
public static class IncrementalCleanupStrategy implements CleanupStrategies.CleanupStrategy {
private static final long serialVersionUID = 3109278696501988780L;
+ static final IncrementalCleanupStrategy DEFAULT_INCREMENTAL_CLEANUP_STRATEGY = new IncrementalCleanupStrategy(5, false);
+
/** Max number of keys pulled from queue for clean up upon state touch for any key. */
private final int cleanupSize;
@@ -407,6 +433,9 @@ public class StateTtlConfig implements Serializable {
public static class RocksdbCompactFilterCleanupStrategy implements CleanupStrategies.CleanupStrategy {
private static final long serialVersionUID = 3109278796506988980L;
+ static final RocksdbCompactFilterCleanupStrategy DEFAULT_ROCKSDB_COMPACT_FILTER_CLEANUP_STRATEGY =
+ new RocksdbCompactFilterCleanupStrategy(1000L);
+
/** Number of state entries to process by compaction filter before updating current timestamp. */
private final long queryTimeAfterNumEntries;
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/state/StateTtlConfigTest.java b/flink-core/src/test/java/org/apache/flink/api/common/state/StateTtlConfigTest.java
new file mode 100644
index 0000000..b78b02b
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/state/StateTtlConfigTest.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.api.common.time.Time;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for the {@link StateTtlConfig}.
+ */
+public class StateTtlConfigTest {
+
+ @Test
+ public void testStateTtlConfigBuildWithCleanupInBackground() {
+ StateTtlConfig ttlConfig = StateTtlConfig
+ .newBuilder(Time.seconds(1))
+ .cleanupInBackground()
+ .build();
+
+ assertNotNull(ttlConfig.getCleanupStrategies());
+
+ StateTtlConfig.CleanupStrategies cleanupStrategies = ttlConfig.getCleanupStrategies();
+ StateTtlConfig.IncrementalCleanupStrategy incrementalCleanupStrategy =
+ cleanupStrategies.getIncrementalCleanupStrategy();
+ StateTtlConfig.RocksdbCompactFilterCleanupStrategy rocksdbCleanupStrategy =
+ cleanupStrategies.getRocksdbCompactFilterCleanupStrategy();
+
+ assertTrue(cleanupStrategies.isCleanupInBackground());
+ assertNotNull(incrementalCleanupStrategy);
+ assertNotNull(rocksdbCleanupStrategy);
+ assertTrue(cleanupStrategies.inRocksdbCompactFilter());
+ assertEquals(5, incrementalCleanupStrategy.getCleanupSize());
+ assertFalse(incrementalCleanupStrategy.runCleanupForEveryRecord());
+ assertEquals(1000, rocksdbCleanupStrategy.getQueryTimeAfterNumEntries());
+ }
+
+}
diff --git a/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/DataStreamStateTTLTestProgram.java b/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/DataStreamStateTTLTestProgram.java
index 1e66e44..5232433 100644
--- a/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/DataStreamStateTTLTestProgram.java
+++ b/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/DataStreamStateTTLTestProgram.java
@@ -58,8 +58,7 @@ public class DataStreamStateTTLTestProgram {
TtlTestConfig config = TtlTestConfig.fromArgs(pt);
StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(config.ttl)
.cleanupFullSnapshot()
- .cleanupIncrementally(5, true)
- .cleanupInRocksdbCompactFilter()
+ .cleanupInBackground()
.build();
env
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/ttl/RocksDBTtlStateTestBase.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/ttl/RocksDBTtlStateTestBase.java
index 41b5780..82e5289 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/ttl/RocksDBTtlStateTestBase.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/ttl/RocksDBTtlStateTestBase.java
@@ -99,7 +99,7 @@ public abstract class RocksDBTtlStateTestBase extends TtlStateTestBase {
}
StateDescriptor<?, ?> stateDesc = initTest(getConfBuilder(TTL)
- .cleanupInRocksdbCompactFilter()
+ .cleanupInBackground()
.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
.build());