You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2021/01/22 01:44:33 UTC
[flink-statefun] 01/02: [hotfix] Refactor StateBinder.bind dispatch
logic to runtime
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit d3112ed68607f082b8036420e7b470de6470b479
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Thu Jan 21 15:07:59 2021 +0800
[hotfix] Refactor StateBinder.bind dispatch logic to runtime
---
.../statefun/flink/core/state/FlinkStateBinder.java | 20 +++++++++++++++-----
.../flink/statefun/sdk/annotations/ForRuntime.java | 2 +-
.../statefun/sdk/state/PersistedStateRegistry.java | 8 +-------
.../flink/statefun/sdk/state/StateBinder.java | 21 ++++-----------------
4 files changed, 21 insertions(+), 30 deletions(-)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkStateBinder.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkStateBinder.java
index 6a0421d..5cd505f 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkStateBinder.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/FlinkStateBinder.java
@@ -44,20 +44,30 @@ public final class FlinkStateBinder extends StateBinder {
}
@Override
- public void bindValue(PersistedValue<?> persistedValue) {
+ public void bind(Object stateObject) {
+ if (stateObject instanceof PersistedValue) {
+ bindValue((PersistedValue<?>) stateObject);
+ } else if (stateObject instanceof PersistedTable) {
+ bindTable((PersistedTable<?, ?>) stateObject);
+ } else if (stateObject instanceof PersistedAppendingBuffer) {
+ bindAppendingBuffer((PersistedAppendingBuffer<?>) stateObject);
+ } else {
+ throw new IllegalArgumentException("Unknown persisted state object " + stateObject);
+ }
+ }
+
+ private void bindValue(PersistedValue<?> persistedValue) {
Accessor<?> accessor = state.createFlinkStateAccessor(functionType, persistedValue);
setAccessorRaw(persistedValue, accessor);
}
- @Override
- public void bindTable(PersistedTable<?, ?> persistedTable) {
+ private void bindTable(PersistedTable<?, ?> persistedTable) {
TableAccessor<?, ?> accessor =
state.createFlinkStateTableAccessor(functionType, persistedTable);
setAccessorRaw(persistedTable, accessor);
}
- @Override
- public void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer) {
+ private void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer) {
AppendingBufferAccessor<?> accessor =
state.createFlinkStateAppendingBufferAccessor(functionType, persistedAppendingBuffer);
setAccessorRaw(persistedAppendingBuffer, accessor);
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/annotations/ForRuntime.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/annotations/ForRuntime.java
index 121c791..2c78f66 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/annotations/ForRuntime.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/annotations/ForRuntime.java
@@ -26,5 +26,5 @@ import java.lang.annotation.Target;
* API with specialized implementation
*/
@Documented
-@Target({ElementType.CONSTRUCTOR, ElementType.METHOD})
+@Target({ElementType.CONSTRUCTOR, ElementType.METHOD, ElementType.TYPE})
public @interface ForRuntime {}
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistry.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistry.java
index bce8364..321a661 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistry.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistry.java
@@ -123,12 +123,6 @@ public final class PersistedStateRegistry {
private static final class NonFaultTolerantStateBinder extends StateBinder {
@Override
- public void bindValue(PersistedValue<?> persistedValue) {}
-
- @Override
- public void bindTable(PersistedTable<?, ?> persistedTable) {}
-
- @Override
- public void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer) {}
+ public void bind(Object stateObject) {}
}
}
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/StateBinder.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/StateBinder.java
index 01560cc..1ce33f2 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/StateBinder.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/StateBinder.java
@@ -18,22 +18,9 @@
package org.apache.flink.statefun.sdk.state;
-public abstract class StateBinder {
- public abstract void bindValue(PersistedValue<?> persistedValue);
-
- public abstract void bindTable(PersistedTable<?, ?> persistedTable);
+import org.apache.flink.statefun.sdk.annotations.ForRuntime;
- public abstract void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer);
-
- public final void bind(Object stateObject) {
- if (stateObject instanceof PersistedValue) {
- bindValue((PersistedValue<?>) stateObject);
- } else if (stateObject instanceof PersistedTable) {
- bindTable((PersistedTable<?, ?>) stateObject);
- } else if (stateObject instanceof PersistedAppendingBuffer) {
- bindAppendingBuffer((PersistedAppendingBuffer<?>) stateObject);
- } else {
- throw new IllegalArgumentException("Unknown persisted state object " + stateObject);
- }
- }
+@ForRuntime
+public abstract class StateBinder {
+ public abstract void bind(Object stateObject);
}