[flink-statefun] branch master updated: [FLINK-17644] Use state expiration

[sj...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sjwiesman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git


The following commit(s) were added to refs/heads/master by this push:
     new dd85b11  [FLINK-17644] Use state expiration
dd85b11 is described below

commit dd85b1184715a4e19aac527584e00f9e17c47aeb
Author: Igal Shilman <ig...@gmail.com>
AuthorDate: Fri May 15 13:28:38 2020 +0200

    [FLINK-17644] Use state expiration
    
    This closes #113
---
 .../statefun/flink/core/state/ExpirationUtil.java  | 63 ++++++++++++++++++++++
 .../statefun/flink/core/state/FlinkState.java      | 29 ++++++----
 2 files changed, 81 insertions(+), 11 deletions(-)

diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/ExpirationUtil.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/ExpirationUtil.java
new file mode 100644
index 0000000..e2fb792
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/ExpirationUtil.java
@@ -0,0 +1,63 @@
+/*
+ * 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.statefun.flink.core.state;
+
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.state.StateTtlConfig.Builder;
+import org.apache.flink.api.common.state.StateTtlConfig.StateVisibility;
+import org.apache.flink.api.common.state.StateTtlConfig.TtlTimeCharacteristic;
+import org.apache.flink.api.common.state.StateTtlConfig.UpdateType;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.statefun.sdk.state.Expiration;
+import org.apache.flink.statefun.sdk.state.Expiration.Mode;
+
+final class ExpirationUtil {
+  private ExpirationUtil() {}
+
+  static void configureStateTtl(StateDescriptor<?, ?> handle, Expiration expiration) {
+    if (expiration.mode() == Mode.NONE) {
+      return;
+    }
+    StateTtlConfig ttlConfig = from(expiration);
+    handle.enableTimeToLive(ttlConfig);
+  }
+
+  private static StateTtlConfig from(Expiration expiration) {
+    final long millis = expiration.duration().toMillis();
+    Builder builder = StateTtlConfig.newBuilder(Time.milliseconds(millis));
+    builder.setTtlTimeCharacteristic(TtlTimeCharacteristic.ProcessingTime);
+    builder.setStateVisibility(StateVisibility.NeverReturnExpired);
+    switch (expiration.mode()) {
+      case AFTER_WRITE:
+        {
+          builder.setUpdateType(UpdateType.OnCreateAndWrite);
+          break;
+        }
+      case AFTER_READ_OR_WRITE:
+        {
+          builder.setUpdateType(UpdateType.OnReadAndWrite);
+          break;
+        }
+      default:
+        throw new IllegalArgumentException("Unknown expiration mode " + expiration.mode());
+    }
+    return builder.build();
+  }
+}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkState.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkState.java
index 98c2a47..b46a3be 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkState.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkState.java
@@ -17,6 +17,8 @@
  */
 package org.apache.flink.statefun.flink.core.state;
 
+import static org.apache.flink.statefun.flink.core.state.ExpirationUtil.configureStateTtl;
+
 import java.util.Objects;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.common.state.ListState;
@@ -63,6 +65,7 @@ public final class FlinkState implements State {
     TypeInformation<T> typeInfo = types.registerType(persistedValue.type());
     String stateName = flinkStateName(functionType, persistedValue.name());
     ValueStateDescriptor<T> descriptor = new ValueStateDescriptor<>(stateName, typeInfo);
+    configureStateTtl(descriptor, persistedValue.expiration());
     ValueState<T> handle = runtimeContext.getState(descriptor);
     return new FlinkValueAccessor<>(handle);
   }
@@ -70,23 +73,27 @@ public final class FlinkState implements State {
   @Override
   public <K, V> TableAccessor<K, V> createFlinkStateTableAccessor(
       FunctionType functionType, PersistedTable<K, V> persistedTable) {
-    MapState<K, V> handle =
-        runtimeContext.getMapState(
-            new MapStateDescriptor<>(
-                flinkStateName(functionType, persistedTable.name()),
-                types.registerType(persistedTable.keyType()),
-                types.registerType(persistedTable.valueType())));
+
+    MapStateDescriptor<K, V> descriptor =
+        new MapStateDescriptor<>(
+            flinkStateName(functionType, persistedTable.name()),
+            types.registerType(persistedTable.keyType()),
+            types.registerType(persistedTable.valueType()));
+
+    configureStateTtl(descriptor, persistedTable.expiration());
+    MapState<K, V> handle = runtimeContext.getMapState(descriptor);
     return new FlinkTableAccessor<>(handle);
   }
 
   @Override
   public <E> AppendingBufferAccessor<E> createFlinkStateAppendingBufferAccessor(
       FunctionType functionType, PersistedAppendingBuffer<E> persistedAppendingBuffer) {
-    ListState<E> handle =
-        runtimeContext.getListState(
-            new ListStateDescriptor<>(
-                flinkStateName(functionType, persistedAppendingBuffer.name()),
-                types.registerType(persistedAppendingBuffer.elementType())));
+    ListStateDescriptor<E> descriptor =
+        new ListStateDescriptor<>(
+            flinkStateName(functionType, persistedAppendingBuffer.name()),
+            types.registerType(persistedAppendingBuffer.elementType()));
+    configureStateTtl(descriptor, persistedAppendingBuffer.expiration());
+    ListState<E> handle = runtimeContext.getListState(descriptor);
     return new FlinkAppendingBufferAccessor<>(handle);
   }