You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2024/03/02 04:41:20 UTC
(flink) branch release-1.19 updated (12ea64c0e2a -> 7618bdeeab0)
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a change to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
from 12ea64c0e2a [FLINK-33436][docs] Add the docs of built-in async-profiler
new 161defe0bb2 [FLINK-34522][core] Changing the Time to Duration for StateTtlConfig.Builder.cleanupInRocksdbCompactFilter
new 697d2b62a5a [hotfix] Fix the StateTtlConfig#newBuilder doc from Time to Duration
new 7618bdeeab0 [Hotfix] Fix Duration class can't load for pyflink
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../docs/dev/datastream/fault-tolerance/state.md | 29 +++++++++++-----------
.../docs/dev/datastream/fault-tolerance/state.md | 29 +++++++++++-----------
.../flink/api/common/state/StateTtlConfig.java | 12 ++++-----
.../flink/api/common/state/StateTtlConfigTest.java | 3 ++-
flink-python/pyflink/datastream/state.py | 14 ++++++-----
.../flink/streaming/api/utils/ProtoUtilsTest.java | 3 ++-
.../state/ttl/RocksDbTtlCompactFiltersManager.java | 2 +-
7 files changed, 49 insertions(+), 43 deletions(-)
(flink) 01/03: [FLINK-34522][core] Changing the Time to Duration for StateTtlConfig.Builder.cleanupInRocksdbCompactFilter
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 161defe0bb2dc8136133e07699b6ac433d52dc65
Author: Rui Fan <19...@gmail.com>
AuthorDate: Tue Feb 27 13:17:25 2024 +0800
[FLINK-34522][core] Changing the Time to Duration for StateTtlConfig.Builder.cleanupInRocksdbCompactFilter
---
docs/content.zh/docs/dev/datastream/fault-tolerance/state.md | 9 +++++----
docs/content/docs/dev/datastream/fault-tolerance/state.md | 9 +++++----
.../org/apache/flink/api/common/state/StateTtlConfig.java | 12 ++++++------
.../apache/flink/api/common/state/StateTtlConfigTest.java | 3 ++-
flink-python/pyflink/datastream/state.py | 8 ++++----
.../org/apache/flink/streaming/api/utils/ProtoUtilsTest.java | 3 ++-
.../streaming/state/ttl/RocksDbTtlCompactFiltersManager.java | 2 +-
7 files changed, 25 insertions(+), 21 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
index c7aaaf0c27f..80f444b13c5 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
@@ -538,7 +538,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter(1000, Time.hours(1))
+ .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build();
```
{{< /tab >}}
@@ -548,18 +548,19 @@ import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter(1000, Time.hours(1))
+ .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build
```
{{< /tab >}}
{{< tab "Python" >}}
```python
+from pyflink.common import Duration
from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfig
ttl_config = StateTtlConfig \
.new_builder(Time.seconds(1)) \
- .cleanup_in_rocksdb_compact_filter(1000, Time.hours(1)) \
+ .cleanup_in_rocksdb_compact_filter(1000, Duration.of_hours(1)) \
.build()
```
{{< /tab >}}
@@ -573,7 +574,7 @@ RocksDB backend 的默认后台清理策略会每处理 1000 条数据进行一
定期压缩可以加速过期状态条目的清理，特别是对于很少访问的状态条目。
比这个值早的文件将被选取进行压缩，并重新写入与之前相同的 Level 中。
该功能可以确保文件定期通过压缩过滤器压缩。
-您可以通过`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Time periodicCompactionTime)`
+您可以通过`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Duration periodicCompactionTime)`
方法设定定期压缩的时间。
定期压缩的时间的默认值是 30 天。
您可以将其设置为 0 以关闭定期压缩或设置一个较小的值以加速过期状态条目的清理，但它将会触发更多压缩。
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state.md b/docs/content/docs/dev/datastream/fault-tolerance/state.md
index 24f14953b1c..0d5705ee196 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/state.md
@@ -601,7 +601,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter(1000, Time.hours(1))
+ .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build();
```
{{< /tab >}}
@@ -611,18 +611,19 @@ import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter(1000, Time.hours(1))
+ .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build
```
{{< /tab >}}
{{< tab "Python" >}}
```python
+from pyflink.common import Duration
from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfig
ttl_config = StateTtlConfig \
.new_builder(Time.seconds(1)) \
- .cleanup_in_rocksdb_compact_filter(1000, Time.hours(1)) \
+ .cleanup_in_rocksdb_compact_filter(1000, Duration.of_hours(1)) \
.build()
```
{{< /tab >}}
@@ -640,7 +641,7 @@ Periodic compaction could speed up expired state entries cleanup, especially for
Files older than this value will be picked up for compaction, and re-written to the same level as they were before.
It makes sure a file goes through compaction filters periodically.
You can change it and pass a custom value to
-`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Time periodicCompactionTime)` method.
+`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Duration periodicCompactionTime)` method.
The default value of Periodic compaction seconds is 30 days.
You could set it to 0 to turn off periodic compaction or set a small value to speed up expired state entries cleanup, but it
would trigger more compactions.
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 26ef85f49e1..d8cc99a5735 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
@@ -341,7 +341,7 @@ public class StateTtlConfig implements Serializable {
*/
@Nonnull
public Builder cleanupInRocksdbCompactFilter(
- long queryTimeAfterNumEntries, Time periodicCompactionTime) {
+ long queryTimeAfterNumEntries, Duration periodicCompactionTime) {
strategies.put(
CleanupStrategies.Strategies.ROCKSDB_COMPACTION_FILTER,
new RocksdbCompactFilterCleanupStrategy(
@@ -354,7 +354,7 @@ public class StateTtlConfig implements Serializable {
*
* <p>If some specific cleanup is configured, e.g. {@link #cleanupIncrementally(int,
* boolean)} or {@link #cleanupInRocksdbCompactFilter(long)} or {@link
- * #cleanupInRocksdbCompactFilter(long, Time)} , this setting does not disable it.
+ * #cleanupInRocksdbCompactFilter(long, Duration)} , this setting does not disable it.
*/
@Nonnull
public Builder disableCleanupInBackground() {
@@ -497,7 +497,7 @@ public class StateTtlConfig implements Serializable {
* Default value is 30 days so that every file goes through the compaction process at least
* once every 30 days if not compacted sooner.
*/
- static final Time DEFAULT_PERIODIC_COMPACTION_TIME = Time.days(30);
+ static final Duration DEFAULT_PERIODIC_COMPACTION_TIME = Duration.ofDays(30);
static final RocksdbCompactFilterCleanupStrategy
DEFAULT_ROCKSDB_COMPACT_FILTER_CLEANUP_STRATEGY =
@@ -515,14 +515,14 @@ public class StateTtlConfig implements Serializable {
* and re-written to the same level as they were before. It makes sure a file goes through
* compaction filters periodically. 0 means turning off periodic compaction.
*/
- private final Time periodicCompactionTime;
+ private final Duration periodicCompactionTime;
private RocksdbCompactFilterCleanupStrategy(long queryTimeAfterNumEntries) {
this(queryTimeAfterNumEntries, DEFAULT_PERIODIC_COMPACTION_TIME);
}
private RocksdbCompactFilterCleanupStrategy(
- long queryTimeAfterNumEntries, Time periodicCompactionTime) {
+ long queryTimeAfterNumEntries, Duration periodicCompactionTime) {
this.queryTimeAfterNumEntries = queryTimeAfterNumEntries;
this.periodicCompactionTime = periodicCompactionTime;
}
@@ -531,7 +531,7 @@ public class StateTtlConfig implements Serializable {
return queryTimeAfterNumEntries;
}
- public Time getPeriodicCompactionTime() {
+ public Duration getPeriodicCompactionTime() {
return periodicCompactionTime;
}
}
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
index a90040c4629..8050ffb7eba 100644
--- 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
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.time.Time;
import org.junit.Assert;
import org.junit.Test;
+import java.time.Duration;
import java.util.Arrays;
import java.util.List;
@@ -74,7 +75,7 @@ public class StateTtlConfigTest {
assertThat(incrementalCleanupStrategy.getCleanupSize(), is(5));
assertThat(incrementalCleanupStrategy.runCleanupForEveryRecord(), is(false));
assertThat(rocksdbCleanupStrategy.getQueryTimeAfterNumEntries(), is(1000L));
- assertThat(rocksdbCleanupStrategy.getPeriodicCompactionTime(), is(Time.days(30)));
+ assertThat(rocksdbCleanupStrategy.getPeriodicCompactionTime(), is(Duration.ofDays(30)));
}
@Test
diff --git a/flink-python/pyflink/datastream/state.py b/flink-python/pyflink/datastream/state.py
index 8450e41df66..a3dc9f6df84 100644
--- a/flink-python/pyflink/datastream/state.py
+++ b/flink-python/pyflink/datastream/state.py
@@ -19,7 +19,7 @@ from abc import ABC, abstractmethod
from enum import Enum
from typing import TypeVar, Generic, Iterable, List, Iterator, Dict, Tuple, Optional
-from pyflink.common.time import Time
+from pyflink.common.time import Duration, Time
from pyflink.common.typeinfo import TypeInformation, Types
__all__ = [
@@ -809,7 +809,7 @@ class StateTtlConfig(object):
def cleanup_in_rocksdb_compact_filter(
self,
query_time_after_num_entries,
- periodic_compaction_time=Time.days(30)) -> \
+ periodic_compaction_time=Duration.of_days(30)) -> \
'StateTtlConfig.Builder':
"""
Cleanup expired state while Rocksdb compaction is running.
@@ -925,14 +925,14 @@ class StateTtlConfig(object):
def __init__(self,
query_time_after_num_entries: int,
- periodic_compaction_time=Time.days(30)):
+ periodic_compaction_time=Duration.of_days(30)):
self._query_time_after_num_entries = query_time_after_num_entries
self._periodic_compaction_time = periodic_compaction_time
def get_query_time_after_num_entries(self) -> int:
return self._query_time_after_num_entries
- def get_periodic_compaction_time(self) -> Time:
+ def get_periodic_compaction_time(self) -> Duration:
return self._periodic_compaction_time
EMPTY_STRATEGY = EmptyCleanupStrategy()
diff --git a/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ProtoUtilsTest.java b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ProtoUtilsTest.java
index 0907896d801..5e81a4b9fef 100644
--- a/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ProtoUtilsTest.java
+++ b/flink-python/src/test/java/org/apache/flink/streaming/api/utils/ProtoUtilsTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.python.util.ProtoUtils;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
@@ -118,6 +119,6 @@ class ProtoUtilsTest {
assertThat(rocksdbCompactFilterCleanupStrategy.getQueryTimeAfterNumEntries())
.isEqualTo(1000);
assertThat(rocksdbCompactFilterCleanupStrategy.getPeriodicCompactionTime())
- .isEqualTo(Time.days(30));
+ .isEqualTo(Duration.ofDays(30));
}
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
index e89fe738a89..168da2f2af1 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
@@ -126,7 +126,7 @@ public class RocksDbTtlCompactFiltersManager {
columnFamilyOptionsMap.get(stateDesc.getName());
Preconditions.checkNotNull(columnFamilyOptions);
columnFamilyOptions.setPeriodicCompactionSeconds(
- rocksdbCompactFilterCleanupStrategy.getPeriodicCompactionTime().toSeconds());
+ rocksdbCompactFilterCleanupStrategy.getPeriodicCompactionTime().getSeconds());
long queryTimeAfterNumEntries =
rocksdbCompactFilterCleanupStrategy.getQueryTimeAfterNumEntries();
(flink) 02/03: [hotfix] Fix the StateTtlConfig#newBuilder doc from Time to Duration
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 697d2b62a5a070ce074550b8767fbc33859db83d
Author: Rui Fan <19...@gmail.com>
AuthorDate: Tue Feb 27 13:55:19 2024 +0800
[hotfix] Fix the StateTtlConfig#newBuilder doc from Time to Duration
---
.../docs/dev/datastream/fault-tolerance/state.md | 20 ++++++++++----------
.../docs/dev/datastream/fault-tolerance/state.md | 20 ++++++++++----------
2 files changed, 20 insertions(+), 20 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
index 80f444b13c5..2635c11f1ec 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
@@ -305,7 +305,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
@@ -321,7 +321,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.common.time.Time
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build
@@ -399,7 +399,7 @@ Heap state backend 会额外存储一个包括用户状态以及时间戳的 Jav
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.disableCleanupInBackground()
.build();
```
@@ -409,7 +409,7 @@ StateTtlConfig ttlConfig = StateTtlConfig
import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.disableCleanupInBackground
.build
```
@@ -441,7 +441,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupFullSnapshot()
.build();
```
@@ -452,7 +452,7 @@ import org.apache.flink.api.common.state.StateTtlConfig
import org.apache.flink.api.common.time.Time
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupFullSnapshot
.build
```
@@ -487,7 +487,7 @@ ttl_config = StateTtlConfig \
```java
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupIncrementally(10, true)
.build();
```
@@ -496,7 +496,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
```scala
import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupIncrementally(10, true)
.build
```
@@ -537,7 +537,7 @@ Flink 提供的 RocksDB 压缩过滤器会在压缩时过滤掉已经过期的
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build();
```
@@ -547,7 +547,7 @@ StateTtlConfig ttlConfig = StateTtlConfig
import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build
```
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state.md b/docs/content/docs/dev/datastream/fault-tolerance/state.md
index 0d5705ee196..233f9bdc01b 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/state.md
@@ -344,7 +344,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
@@ -360,7 +360,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.common.time.Time
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build
@@ -447,7 +447,7 @@ in the background if supported by the configured state backend. Background clean
```java
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.disableCleanupInBackground()
.build();
```
@@ -456,7 +456,7 @@ StateTtlConfig ttlConfig = StateTtlConfig
```scala
import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.disableCleanupInBackground
.build
```
@@ -491,7 +491,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupFullSnapshot()
.build();
```
@@ -502,7 +502,7 @@ import org.apache.flink.api.common.state.StateTtlConfig
import org.apache.flink.api.common.time.Time
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupFullSnapshot
.build
```
@@ -543,7 +543,7 @@ This feature can be configured in `StateTtlConfig`:
```java
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupIncrementally(10, true)
.build();
```
@@ -552,7 +552,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
```scala
import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupIncrementally(10, true)
.build
```
@@ -600,7 +600,7 @@ This feature can be configured in `StateTtlConfig`:
import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build();
```
@@ -610,7 +610,7 @@ StateTtlConfig ttlConfig = StateTtlConfig
import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
- .newBuilder(Time.seconds(1))
+ .newBuilder(Duration.ofSeconds(1))
.cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
.build
```
(flink) 03/03: [Hotfix] Fix Duration class can't load for pyflink
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7618bdeeab06c09219136a04a62262148c677134
Author: Roc Marshal <fl...@126.com>
AuthorDate: Fri Mar 1 09:57:48 2024 +0800
[Hotfix] Fix Duration class can't load for pyflink
---
flink-python/pyflink/datastream/state.py | 10 ++++++----
1 file changed, 6 insertions(+), 4 deletions(-)
diff --git a/flink-python/pyflink/datastream/state.py b/flink-python/pyflink/datastream/state.py
index a3dc9f6df84..4216561299c 100644
--- a/flink-python/pyflink/datastream/state.py
+++ b/flink-python/pyflink/datastream/state.py
@@ -809,7 +809,7 @@ class StateTtlConfig(object):
def cleanup_in_rocksdb_compact_filter(
self,
query_time_after_num_entries,
- periodic_compaction_time=Duration.of_days(30)) -> \
+ periodic_compaction_time=None) -> \
'StateTtlConfig.Builder':
"""
Cleanup expired state while Rocksdb compaction is running.
@@ -833,7 +833,8 @@ class StateTtlConfig(object):
self._strategies[
StateTtlConfig.CleanupStrategies.Strategies.ROCKSDB_COMPACTION_FILTER] = \
StateTtlConfig.CleanupStrategies.RocksdbCompactFilterCleanupStrategy(
- query_time_after_num_entries, periodic_compaction_time)
+ query_time_after_num_entries,
+ periodic_compaction_time if periodic_compaction_time else Duration.of_days(30))
return self
def disable_cleanup_in_background(self) -> 'StateTtlConfig.Builder':
@@ -925,9 +926,10 @@ class StateTtlConfig(object):
def __init__(self,
query_time_after_num_entries: int,
- periodic_compaction_time=Duration.of_days(30)):
+ periodic_compaction_time=None):
self._query_time_after_num_entries = query_time_after_num_entries
- self._periodic_compaction_time = periodic_compaction_time
+ self._periodic_compaction_time = periodic_compaction_time \
+ if periodic_compaction_time else Duration.of_days(30)
def get_query_time_after_num_entries(self) -> int:
return self._query_time_after_num_entries