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 2020/09/01 07:38:32 UTC
[flink-statefun] branch master updated (9a76143 -> 822c6e6)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git.
from 9a76143 [FLINK-19095] [core] Respect state expiration configuration in PersistedRemoteFunctionValues
new f26df15 [FLINK-19096] [sdk] Rework PersistedStateRegistry registration methods
new dd095c8 [FLINK-19096] [sdk] Add toString() for state classes for better error messages
new bda7fa6 [FLINK-19102] [core, sdk] Make StateBinder a per-FunctionType entity
new 822c6e6 [hotfix] [core] Rename PersistedStates.findReflectivelyAndBind
The 4 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:
.../statefun/flink/core/functions/Reductions.java | 2 -
.../core/functions/StatefulFunctionRepository.java | 12 +-
.../reqreply/PersistedRemoteFunctionValues.java | 11 +-
.../flink/core/state/FlinkStateBinder.java | 16 +-
.../statefun/flink/core/state/PersistedStates.java | 9 +-
.../flink/statefun/sdk/state/ApiExtension.java | 8 +-
.../flink/core/state/PersistedStatesTest.java | 40 ++---
.../operator/StateBootstrapFunctionRegistry.java | 10 +-
.../flink/statefun/sdk/state/Expiration.java | 5 +
.../sdk/state/PersistedAppendingBuffer.java | 7 +
.../statefun/sdk/state/PersistedStateRegistry.java | 190 +++++----------------
.../flink/statefun/sdk/state/PersistedTable.java | 7 +
.../flink/statefun/sdk/state/PersistedValue.java | 6 +
.../flink/statefun/sdk/state/StateBinder.java | 17 +-
.../sdk/state/PersistedStateRegistryTest.java | 12 +-
15 files changed, 133 insertions(+), 219 deletions(-)
[flink-statefun] 02/04: [FLINK-19096] [sdk] Add toString() for
state classes for better error messages
Posted by tz...@apache.org.
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 dd095c8236f7e0a0b8548ea3dc6b40b5728210d8
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Aug 31 15:59:46 2020 +0800
[FLINK-19096] [sdk] Add toString() for state classes for better error messages
This closes #136.
---
.../main/java/org/apache/flink/statefun/sdk/state/Expiration.java | 5 +++++
.../apache/flink/statefun/sdk/state/PersistedAppendingBuffer.java | 7 +++++++
.../apache/flink/statefun/sdk/state/PersistedStateRegistry.java | 4 +---
.../java/org/apache/flink/statefun/sdk/state/PersistedTable.java | 7 +++++++
.../java/org/apache/flink/statefun/sdk/state/PersistedValue.java | 6 ++++++
5 files changed, 26 insertions(+), 3 deletions(-)
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/Expiration.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/Expiration.java
index d3396f3..8841cc9 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/Expiration.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/Expiration.java
@@ -86,4 +86,9 @@ public final class Expiration implements Serializable {
public Duration duration() {
return duration;
}
+
+ @Override
+ public String toString() {
+ return String.format("Expiration{mode=%s, duration=%s}", mode, duration);
+ }
}
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedAppendingBuffer.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedAppendingBuffer.java
index 42032c5..60541ff 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedAppendingBuffer.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedAppendingBuffer.java
@@ -160,6 +160,13 @@ public final class PersistedAppendingBuffer<E> {
accessor.clear();
}
+ @Override
+ public String toString() {
+ return String.format(
+ "PersistedAppendingBuffer{name=%s, elementType=%s, expiration=%s}",
+ name, elementType.getName(), expiration);
+ }
+
@ForRuntime
void setAccessor(AppendingBufferAccessor<E> newAccessor) {
this.accessor = Objects.requireNonNull(newAccessor);
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 f606675..b806df1 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
@@ -121,9 +121,7 @@ public final class PersistedStateRegistry {
throw new IllegalStateException(
String.format(
"State name '%s' was registered twice; previous registered state object with the same name was a %s, attempting to register a new %s under the same name.",
- stateName,
- previousRegistration.getClass().getName(),
- newStateObject.getClass().getName()));
+ stateName, previousRegistration, newStateObject));
}
registeredStates.put(stateName, newStateObject);
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java
index a04c126..4826606 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java
@@ -182,6 +182,13 @@ public final class PersistedTable<K, V> {
accessor.clear();
}
+ @Override
+ public String toString() {
+ return String.format(
+ "PersistedTable{name=%s, keyType=%s, valueType=%s, expiration=%s}",
+ name, keyType.getName(), valueType.getName(), expiration);
+ }
+
@ForRuntime
void setAccessor(TableAccessor<K, V> newAccessor) {
Objects.requireNonNull(newAccessor);
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedValue.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedValue.java
index 3d78b75..7c29493 100644
--- a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedValue.java
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedValue.java
@@ -165,6 +165,12 @@ public final class PersistedValue<T> {
this.accessor = newAccessor;
}
+ @Override
+ public String toString() {
+ return String.format(
+ "PersistedValue{name=%s, type=%s, expiration=%s}", name, type.getName(), expiration);
+ }
+
private static final class NonFaultTolerantAccessor<E> implements Accessor<E> {
private E element;
[flink-statefun] 01/04: [FLINK-19096] [sdk] Rework
PersistedStateRegistry registration methods
Posted by tz...@apache.org.
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 f26df15fc91a9bb544dfc77f9a089d5952a5f76c
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Aug 31 15:48:39 2020 +0800
[FLINK-19096] [sdk] Rework PersistedStateRegistry registration methods
This reworks the `registerValue` / `registerTable` etc. methods on
PersistedStateRegistry to directly accept a constructed PersistedValue /
PersistedTable instance. This allows us to avoid having to synchronize
those methods and the state class constructors.
---
.../reqreply/PersistedRemoteFunctionValues.java | 11 +-
.../flink/core/state/PersistedStatesTest.java | 16 ++-
.../statefun/sdk/state/PersistedStateRegistry.java | 160 ++++-----------------
.../sdk/state/PersistedStateRegistryTest.java | 12 +-
4 files changed, 53 insertions(+), 146 deletions(-)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/reqreply/PersistedRemoteFunctionValues.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/reqreply/PersistedRemoteFunctionValues.java
index e918c41..4733366 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/reqreply/PersistedRemoteFunctionValues.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/reqreply/PersistedRemoteFunctionValues.java
@@ -37,7 +37,7 @@ public final class PersistedRemoteFunctionValues {
public PersistedRemoteFunctionValues(List<StateSpec> stateSpecs) {
Objects.requireNonNull(stateSpecs);
this.managedStates = new HashMap<>(stateSpecs.size());
- stateSpecs.forEach(spec -> managedStates.put(spec.name(), createStateHandle(spec)));
+ stateSpecs.forEach(this::createAndRegisterValueState);
}
void forEach(BiConsumer<String, byte[]> consumer) {
@@ -52,8 +52,13 @@ public final class PersistedRemoteFunctionValues {
getStateHandleOrThrow(stateName).clear();
}
- private PersistedValue<byte[]> createStateHandle(StateSpec stateSpec) {
- return stateRegistry.registerValue(stateSpec.name(), byte[].class, stateSpec.ttlExpiration());
+ private void createAndRegisterValueState(StateSpec stateSpec) {
+ final String stateName = stateSpec.name();
+
+ final PersistedValue<byte[]> stateValue =
+ PersistedValue.of(stateName, byte[].class, stateSpec.ttlExpiration());
+ stateRegistry.registerValue(stateValue);
+ managedStates.put(stateName, stateValue);
}
private PersistedValue<byte[]> getStateHandleOrThrow(String stateName) {
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
index 3e1c622..5e5d4fb 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
@@ -180,15 +180,19 @@ public class PersistedStatesTest {
@Persisted PersistedStateRegistry provider = new PersistedStateRegistry();
DynamicState() {
- provider.registerValue("in-constructor-value", String.class);
- provider.registerTable("in-constructor-table", String.class, Integer.class);
- provider.registerAppendingBuffer("in-constructor-buffer", String.class);
+ provider.registerValue(PersistedValue.of("in-constructor-value", String.class));
+ provider.registerTable(
+ PersistedTable.of("in-constructor-table", String.class, Integer.class));
+ provider.registerAppendingBuffer(
+ PersistedAppendingBuffer.of("in-constructor-buffer", String.class));
}
void process() {
- provider.registerValue("post-constructor-value", String.class);
- provider.registerTable("post-constructor-table", String.class, Integer.class);
- provider.registerAppendingBuffer("post-constructor-buffer", String.class);
+ provider.registerValue(PersistedValue.of("post-constructor-value", String.class));
+ provider.registerTable(
+ PersistedTable.of("post-constructor-table", String.class, Integer.class));
+ provider.registerAppendingBuffer(
+ PersistedAppendingBuffer.of("post-constructor-buffer", String.class));
}
}
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 391bffe..f606675 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
@@ -21,7 +21,6 @@ package org.apache.flink.statefun.sdk.state;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.function.Function;
import javax.annotation.Nullable;
import org.apache.flink.statefun.sdk.FunctionType;
import org.apache.flink.statefun.sdk.StatefulFunction;
@@ -56,121 +55,40 @@ public final class PersistedStateRegistry {
}
/**
- * Registers a {@link PersistedValue}, given a state name and the type of the values. If a
- * registered value already exists for the given name, the previous persisted value is returned.
+ * Registers a {@link PersistedValue}. If a registered state already exists for the specified name
+ * of the value, the registration fails.
*
- * @param name the state name to register with.
- * @param type the type of the value.
+ * @param valueState the value state to register.
* @param <T> the type of the value.
- * @return the registered value, or the previous registered value if a registration for the state
- * name already exists.
- * @throws IllegalStateException if a previous registration exists for the given state name, but
- * it wasn't registered as a {@link PersistedValue}.
+ * @throws IllegalStateException if a previous registration exists for the given state name.
*/
- public <T> PersistedValue<T> registerValue(String name, Class<T> type) {
- return registerValue(name, type, Expiration.none());
+ public <T> void registerValue(PersistedValue<T> valueState) {
+ acceptRegistrationOrThrowIfPresent(valueState.name(), valueState);
}
/**
- * Registers a {@link PersistedValue}, given a state name and the type of the values. If a
- * registered value already exists for the given name, the previous persisted value is returned.
+ * Registers a {@link PersistedTable}. If a registered state already exists for the specified name
+ * of the table, the registration fails.
*
- * @param name the state name to register with.
- * @param type the type of the value.
- * @param expiration expiration configuration for the registered state.
- * @param <T> the type of the value.
- * @return the registered value, or the previous registered value if a registration for the state
- * name already exists.
- * @throws IllegalStateException if a previous registration exists for the given state name, but
- * it wasn't registered as a {@link PersistedValue}.
- */
- public <T> PersistedValue<T> registerValue(String name, Class<T> type, Expiration expiration) {
- return getStateOrCreateIfAbsent(
- PersistedValue.class, name, stateName -> createValue(stateName, type, expiration));
- }
-
- /**
- * Registers a {@link PersistedTable}, given a state name and the type of the keys and values of
- * the table. If a registered value already exists for the given name, the previous persisted
- * table is returned.
- *
- * @param name the state name to register with.
- * @param keyType the type of the keys.
- * @param valueType the type of the values.
- * @param <K> the type of the keys.
- * @param <V> the type of the values.
- * @return the registered table, or the previous registered table if a registration for the state
- * name already exists.
- * @throws IllegalStateException if a previous registration exists for the given state name, but
- * it wasn't registered as a {@link PersistedTable}.
- */
- public <K, V> PersistedTable<K, V> registerTable(
- String name, Class<K> keyType, Class<V> valueType) {
- return registerTable(name, keyType, valueType, Expiration.none());
- }
-
- /**
- * Registers a {@link PersistedTable}, given a state name and the type of the keys and values of
- * the table. If a registered value already exists for the given name, the previous persisted
- * table is returned.
- *
- * @param name the state name to register with.
- * @param keyType the type of the keys.
- * @param valueType the type of the values.
- * @param expiration expiration configuration for the registered state.
+ * @param tableState the table state to register.
* @param <K> the type of the keys.
* @param <V> the type of the values.
- * @return the registered table, or the previous registered table if a registration for the state
- * name already exists.
- * @throws IllegalStateException if a previous registration exists for the given state name, but
- * it wasn't registered as a {@link PersistedTable}.
+ * @throws IllegalStateException if a previous registration exists for the given state name.
*/
- public <K, V> PersistedTable<K, V> registerTable(
- String name, Class<K> keyType, Class<V> valueType, Expiration expiration) {
- return getStateOrCreateIfAbsent(
- PersistedTable.class,
- name,
- stateName -> createTable(stateName, keyType, valueType, expiration));
+ public <K, V> void registerTable(PersistedTable<K, V> tableState) {
+ acceptRegistrationOrThrowIfPresent(tableState.name(), tableState);
}
/**
- * Registers a {@link PersistedAppendingBuffer}, given a state name and the type of the buffer
- * elements. If a registered buffer already exists for the given name, the previous persisted
- * buffer is returned.
+ * Registers a {@link PersistedAppendingBuffer}. If a registered state already exists for the
+ * specified name of the table, the registration fails.
*
- * @param name the state name to register with.
- * @param elementType the type of the buffer elements.
+ * @param bufferState the appending buffer to register.
* @param <E> the type of the buffer elements.
- * @return the registered buffer, or the previous registered buffer if a registration for the
- * state name already exists.
- * @throws IllegalStateException if a previous registration exists for the given state name, but
- * it wasn't registered as a {@link PersistedAppendingBuffer}.
+ * @throws IllegalStateException if a previous registration exists for the given state name.
*/
- public <E> PersistedAppendingBuffer<E> registerAppendingBuffer(
- String name, Class<E> elementType) {
- return registerAppendingBuffer(name, elementType, Expiration.none());
- }
-
- /**
- * Registers a {@link PersistedAppendingBuffer}, given a state name and the type of the buffer
- * elements. If a registered buffer already exists for the given name, the previous persisted
- * buffer is returned.
- *
- * @param name the state name to register with.
- * @param elementType the type of the buffer elements.
- * @param expiration expiration configuration for the registered state.
- * @param <E> the type of the buffer elements.
- * @return the registered buffer, or the previous registered buffer if a registration for the
- * state name already exists.
- * @throws IllegalStateException if a previous registration exists for the given state name, but
- * it wasn't registered as a {@link PersistedAppendingBuffer}.
- */
- public <E> PersistedAppendingBuffer<E> registerAppendingBuffer(
- String name, Class<E> elementType, Expiration expiration) {
- return getStateOrCreateIfAbsent(
- PersistedAppendingBuffer.class,
- name,
- stateName -> createAppendingBuffer(stateName, elementType, expiration));
+ public <E> void registerAppendingBuffer(PersistedAppendingBuffer<E> bufferState) {
+ acceptRegistrationOrThrowIfPresent(bufferState.name(), bufferState);
}
/**
@@ -197,39 +115,19 @@ public final class PersistedStateRegistry {
registeredStates.values().forEach(state -> stateBinder.bind(state, functionType));
}
- private <T> PersistedValue<T> createValue(String name, Class<T> type, Expiration expiration) {
- final PersistedValue<T> value = PersistedValue.of(name, type, expiration);
- stateBinder.bindValue(value, functionType);
- return value;
- }
-
- private <K, V> PersistedTable<K, V> createTable(
- String name, Class<K> keyType, Class<V> valueType, Expiration expiration) {
- final PersistedTable<K, V> table = PersistedTable.of(name, keyType, valueType, expiration);
- stateBinder.bindTable(table, functionType);
- return table;
- }
-
- private <E> PersistedAppendingBuffer<E> createAppendingBuffer(
- String name, Class<E> elementType, Expiration expiration) {
- final PersistedAppendingBuffer<E> buffer =
- PersistedAppendingBuffer.of(name, elementType, expiration);
- stateBinder.bindAppendingBuffer(buffer, functionType);
- return buffer;
- }
-
- @SuppressWarnings("unchecked")
- private <ST> ST getStateOrCreateIfAbsent(
- Class<?> statePrimitiveType, String name, Function<String, ST> createFunction) {
- final ST state = (ST) registeredStates.computeIfAbsent(name, createFunction::apply);
- if (state.getClass() != statePrimitiveType) {
+ private void acceptRegistrationOrThrowIfPresent(String stateName, Object newStateObject) {
+ final Object previousRegistration = registeredStates.get(stateName);
+ if (previousRegistration != null) {
throw new IllegalStateException(
- "Unexpected state primitive type. The state was registered with type: "
- + state.getClass()
- + ", but was attempting to access it again as type: "
- + statePrimitiveType);
+ String.format(
+ "State name '%s' was registered twice; previous registered state object with the same name was a %s, attempting to register a new %s under the same name.",
+ stateName,
+ previousRegistration.getClass().getName(),
+ newStateObject.getClass().getName()));
}
- return state;
+
+ registeredStates.put(stateName, newStateObject);
+ stateBinder.bind(newStateObject, functionType);
}
private static final class NonFaultTolerantStateBinder extends StateBinder {
diff --git a/statefun-sdk/src/test/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistryTest.java b/statefun-sdk/src/test/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistryTest.java
index 0c91178..7858b6b 100644
--- a/statefun-sdk/src/test/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistryTest.java
+++ b/statefun-sdk/src/test/java/org/apache/flink/statefun/sdk/state/PersistedStateRegistryTest.java
@@ -26,16 +26,16 @@ public class PersistedStateRegistryTest {
public void exampleUsage() {
final PersistedStateRegistry registry = new PersistedStateRegistry();
- registry.registerValue("value", String.class);
- registry.registerTable("table", String.class, Integer.class);
- registry.registerAppendingBuffer("buffer", String.class);
+ registry.registerValue(PersistedValue.of("value", String.class));
+ registry.registerTable(PersistedTable.of("table", String.class, Integer.class));
+ registry.registerAppendingBuffer(PersistedAppendingBuffer.of("buffer", String.class));
}
@Test(expected = IllegalStateException.class)
- public void reaccessAsWrongStatePrimitiveType() {
+ public void duplicateRegistration() {
final PersistedStateRegistry registry = new PersistedStateRegistry();
- registry.registerValue("my-state", String.class);
- registry.registerAppendingBuffer("my-state", String.class);
+ registry.registerValue(PersistedValue.of("my-state", String.class));
+ registry.registerTable(PersistedTable.of("my-state", String.class, Integer.class));
}
}
[flink-statefun] 03/04: [FLINK-19102] [core,
sdk] Make StateBinder a per-FunctionType entity
Posted by tz...@apache.org.
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 bda7fa6be426d091fc3da04ce31e2f9d8aafb1a7
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Aug 31 16:53:51 2020 +0800
[FLINK-19102] [core, sdk] Make StateBinder a per-FunctionType entity
The goal of this commit is to make StateBinders know the FunctionType,
with each state binder instance specifically binding state for a single
FunctionType.
The side effect of this is also moving FunctionType out of
PersistedStateRegistry, as that information is encapsulated to the
StateBinder it receives at runtime.
This closes #137.
---
.../statefun/flink/core/functions/Reductions.java | 2 --
.../core/functions/StatefulFunctionRepository.java | 12 +++++---
.../flink/core/state/FlinkStateBinder.java | 16 ++++++----
.../statefun/flink/core/state/PersistedStates.java | 8 ++---
.../flink/statefun/sdk/state/ApiExtension.java | 8 ++---
.../flink/core/state/PersistedStatesTest.java | 24 +++++++--------
.../operator/StateBootstrapFunctionRegistry.java | 10 +++---
.../statefun/sdk/state/PersistedStateRegistry.java | 36 +++++++++-------------
.../flink/statefun/sdk/state/StateBinder.java | 17 +++++-----
9 files changed, 58 insertions(+), 75 deletions(-)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/Reductions.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/Reductions.java
index 9c42cdf..abce51a 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/Reductions.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/Reductions.java
@@ -35,7 +35,6 @@ import org.apache.flink.statefun.flink.core.message.MessageFactory;
import org.apache.flink.statefun.flink.core.metrics.FlinkMetricsFactory;
import org.apache.flink.statefun.flink.core.metrics.MetricsFactory;
import org.apache.flink.statefun.flink.core.state.FlinkState;
-import org.apache.flink.statefun.flink.core.state.FlinkStateBinder;
import org.apache.flink.statefun.flink.core.state.State;
import org.apache.flink.statefun.flink.core.types.DynamicallyRegisteredTypes;
import org.apache.flink.statefun.sdk.io.EgressIdentifier;
@@ -94,7 +93,6 @@ final class Reductions {
container.add("applying-context", ApplyingContext.class, ReusableContext.class);
container.add(LocalSink.class);
container.add("function-loader", FunctionLoader.class, PredefinedFunctionLoader.class);
- container.add(FlinkStateBinder.class);
container.add(Reductions.class);
container.add(LocalFunctionGroup.class);
container.add("metrics-factory", MetricsFactory.class, new FlinkMetricsFactory(metricGroup));
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java
index 0187e57..af512d0 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java
@@ -27,11 +27,12 @@ import org.apache.flink.statefun.flink.core.metrics.FunctionTypeMetrics;
import org.apache.flink.statefun.flink.core.metrics.MetricsFactory;
import org.apache.flink.statefun.flink.core.state.FlinkStateBinder;
import org.apache.flink.statefun.flink.core.state.PersistedStates;
+import org.apache.flink.statefun.flink.core.state.State;
import org.apache.flink.statefun.sdk.FunctionType;
final class StatefulFunctionRepository implements FunctionRepository {
private final ObjectOpenHashMap<FunctionType, StatefulFunction> instances;
- private final FlinkStateBinder stateBinder;
+ private final State flinkState;
private final FunctionLoader functionLoader;
private final MetricsFactory metricsFactory;
private final MessageFactory messageFactory;
@@ -40,12 +41,12 @@ final class StatefulFunctionRepository implements FunctionRepository {
StatefulFunctionRepository(
@Label("function-loader") FunctionLoader functionLoader,
@Label("metrics-factory") MetricsFactory metricsFactory,
- MessageFactory messageFactory,
- FlinkStateBinder stateBinder) {
+ @Label("state") State state,
+ MessageFactory messageFactory) {
this.instances = new ObjectOpenHashMap<>();
- this.stateBinder = Objects.requireNonNull(stateBinder);
this.functionLoader = Objects.requireNonNull(functionLoader);
this.metricsFactory = Objects.requireNonNull(metricsFactory);
+ this.flinkState = Objects.requireNonNull(state);
this.messageFactory = Objects.requireNonNull(messageFactory);
}
@@ -62,7 +63,8 @@ final class StatefulFunctionRepository implements FunctionRepository {
org.apache.flink.statefun.sdk.StatefulFunction statefulFunction =
functionLoader.load(functionType);
try (SetContextClassLoader ignored = new SetContextClassLoader(statefulFunction)) {
- PersistedStates.findAndBind(functionType, statefulFunction, stateBinder);
+ FlinkStateBinder stateBinderForType = new FlinkStateBinder(flinkState, functionType);
+ PersistedStates.findAndBind(statefulFunction, stateBinderForType);
FunctionTypeMetrics metrics = metricsFactory.forType(functionType);
return new StatefulFunction(statefulFunction, metrics, messageFactory);
}
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 839e601..6a0421d 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
@@ -19,7 +19,6 @@ package org.apache.flink.statefun.flink.core.state;
import java.util.Objects;
import org.apache.flink.statefun.flink.core.di.Inject;
-import org.apache.flink.statefun.flink.core.di.Label;
import org.apache.flink.statefun.sdk.FunctionType;
import org.apache.flink.statefun.sdk.state.Accessor;
import org.apache.flink.statefun.sdk.state.ApiExtension;
@@ -30,30 +29,35 @@ import org.apache.flink.statefun.sdk.state.PersistedValue;
import org.apache.flink.statefun.sdk.state.StateBinder;
import org.apache.flink.statefun.sdk.state.TableAccessor;
+/**
+ * A {@link StateBinder} that binds persisted state objects to Flink state for a specific {@link
+ * FunctionType}.
+ */
public final class FlinkStateBinder extends StateBinder {
private final State state;
+ private final FunctionType functionType;
@Inject
- public FlinkStateBinder(@Label("state") State state) {
+ public FlinkStateBinder(State state, FunctionType functionType) {
this.state = Objects.requireNonNull(state);
+ this.functionType = Objects.requireNonNull(functionType);
}
@Override
- public void bindValue(PersistedValue<?> persistedValue, FunctionType functionType) {
+ public void bindValue(PersistedValue<?> persistedValue) {
Accessor<?> accessor = state.createFlinkStateAccessor(functionType, persistedValue);
setAccessorRaw(persistedValue, accessor);
}
@Override
- public void bindTable(PersistedTable<?, ?> persistedTable, FunctionType functionType) {
+ public void bindTable(PersistedTable<?, ?> persistedTable) {
TableAccessor<?, ?> accessor =
state.createFlinkStateTableAccessor(functionType, persistedTable);
setAccessorRaw(persistedTable, accessor);
}
@Override
- public void bindAppendingBuffer(
- PersistedAppendingBuffer<?> persistedAppendingBuffer, FunctionType functionType) {
+ public void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer) {
AppendingBufferAccessor<?> accessor =
state.createFlinkStateAppendingBufferAccessor(functionType, persistedAppendingBuffer);
setAccessorRaw(persistedAppendingBuffer, accessor);
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java
index 6e57c5b..968e4b5 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java
@@ -26,7 +26,6 @@ import java.util.List;
import java.util.stream.Stream;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
-import org.apache.flink.statefun.sdk.FunctionType;
import org.apache.flink.statefun.sdk.annotations.Persisted;
import org.apache.flink.statefun.sdk.state.ApiExtension;
import org.apache.flink.statefun.sdk.state.PersistedAppendingBuffer;
@@ -36,15 +35,14 @@ import org.apache.flink.statefun.sdk.state.PersistedValue;
public final class PersistedStates {
- public static void findAndBind(
- FunctionType functionType, @Nullable Object instance, FlinkStateBinder stateBinder) {
+ public static void findAndBind(@Nullable Object instance, FlinkStateBinder stateBinder) {
List<?> states = findReflectively(instance);
for (Object persisted : states) {
if (persisted instanceof PersistedStateRegistry) {
PersistedStateRegistry stateRegistry = (PersistedStateRegistry) persisted;
- ApiExtension.bindPersistedStateRegistry(stateRegistry, stateBinder, functionType);
+ ApiExtension.bindPersistedStateRegistry(stateRegistry, stateBinder);
} else {
- stateBinder.bind(persisted, functionType);
+ stateBinder.bind(persisted);
}
}
}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/sdk/state/ApiExtension.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/sdk/state/ApiExtension.java
index 30e9657..f107aeb 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/sdk/state/ApiExtension.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/sdk/state/ApiExtension.java
@@ -17,8 +17,6 @@
*/
package org.apache.flink.statefun.sdk.state;
-import org.apache.flink.statefun.sdk.FunctionType;
-
public class ApiExtension {
public static <T> void setPersistedValueAccessor(
PersistedValue<T> persistedValue, Accessor<T> accessor) {
@@ -36,9 +34,7 @@ public class ApiExtension {
}
public static void bindPersistedStateRegistry(
- PersistedStateRegistry persistedStateRegistry,
- StateBinder stateBinder,
- FunctionType functionType) {
- persistedStateRegistry.bind(stateBinder, functionType);
+ PersistedStateRegistry persistedStateRegistry, StateBinder stateBinder) {
+ persistedStateRegistry.bind(stateBinder);
}
}
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
index 5e5d4fb..adddcf5 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
@@ -47,52 +47,50 @@ public class PersistedStatesTest {
private final FakeState state = new FakeState();
// object under test
- private final FlinkStateBinder binderUnderTest = new FlinkStateBinder(state);
+ private final FlinkStateBinder binderUnderTest =
+ new FlinkStateBinder(state, TestUtils.FUNCTION_TYPE);
@Test
public void exampleUsage() {
- PersistedStates.findAndBind(TestUtils.FUNCTION_TYPE, new SanityClass(), binderUnderTest);
+ PersistedStates.findAndBind(new SanityClass(), binderUnderTest);
assertThat(state.boundNames, hasItems("name", "last"));
}
@Test(expected = IllegalStateException.class)
public void nullValueField() {
- PersistedStates.findAndBind(TestUtils.FUNCTION_TYPE, new NullValueClass(), binderUnderTest);
+ PersistedStates.findAndBind(new NullValueClass(), binderUnderTest);
}
@Test
public void nonAnnotatedClass() {
- PersistedStates.findAndBind(TestUtils.FUNCTION_TYPE, new IgnoreNonAnnotated(), binderUnderTest);
+ PersistedStates.findAndBind(new IgnoreNonAnnotated(), binderUnderTest);
assertTrue(state.boundNames.isEmpty());
}
@Test
public void extendedClass() {
- PersistedStates.findAndBind(TestUtils.FUNCTION_TYPE, new ChildClass(), binderUnderTest);
+ PersistedStates.findAndBind(new ChildClass(), binderUnderTest);
assertThat(state.boundNames, hasItems("parent", "child"));
}
@Test(expected = IllegalArgumentException.class)
public void staticPersistedFieldsAreNotAllowed() {
- PersistedStates.findAndBind(
- TestUtils.FUNCTION_TYPE, new StaticPersistedValue(), binderUnderTest);
+ PersistedStates.findAndBind(new StaticPersistedValue(), binderUnderTest);
}
@Test
public void bindPersistedTable() {
- PersistedStates.findAndBind(
- TestUtils.FUNCTION_TYPE, new PersistedTableValue(), binderUnderTest);
+ PersistedStates.findAndBind(new PersistedTableValue(), binderUnderTest);
assertThat(state.boundNames, hasItems("table"));
}
@Test
public void bindPersistedAppendingBuffer() {
- PersistedStates.findAndBind(
- TestUtils.FUNCTION_TYPE, new PersistedAppendingBufferState(), binderUnderTest);
+ PersistedStates.findAndBind(new PersistedAppendingBufferState(), binderUnderTest);
assertThat(state.boundNames, hasItems("buffer"));
}
@@ -100,7 +98,7 @@ public class PersistedStatesTest {
@Test
public void bindDynamicState() {
DynamicState dynamicState = new DynamicState();
- PersistedStates.findAndBind(TestUtils.FUNCTION_TYPE, dynamicState, binderUnderTest);
+ PersistedStates.findAndBind(dynamicState, binderUnderTest);
dynamicState.process();
@@ -117,7 +115,7 @@ public class PersistedStatesTest {
@Test
public void bindComposedState() {
- PersistedStates.findAndBind(TestUtils.FUNCTION_TYPE, new OuterClass(), binderUnderTest);
+ PersistedStates.findAndBind(new OuterClass(), binderUnderTest);
assertThat(state.boundNames, hasItems("inner"));
}
diff --git a/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java b/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java
index 304a4df..2825f75 100644
--- a/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java
+++ b/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java
@@ -93,14 +93,14 @@ public final class StateBootstrapFunctionRegistry implements Serializable {
void initialize(State stateAccessor) {
this.registry = new HashMap<>(stateBootstrapFunctionProviders.size());
- final FlinkStateBinder stateBinder = new FlinkStateBinder(stateAccessor);
for (Map.Entry<SerializableFunctionType, StateBootstrapFunctionProvider> entry :
stateBootstrapFunctionProviders.entrySet()) {
final FunctionType functionType = entry.getKey().toNonSerializable();
final StateBootstrapFunction bootstrapFunction =
entry.getValue().bootstrapFunctionOfType(functionType);
+ final FlinkStateBinder stateBinder = new FlinkStateBinder(stateAccessor, functionType);
- registry.put(functionType, bindState(functionType, bootstrapFunction, stateBinder));
+ registry.put(functionType, bindState(bootstrapFunction, stateBinder));
}
}
@@ -115,11 +115,9 @@ public final class StateBootstrapFunctionRegistry implements Serializable {
}
private static StateBootstrapFunction bindState(
- FunctionType functionType,
- StateBootstrapFunction bootstrapFunction,
- FlinkStateBinder stateBinder) {
+ StateBootstrapFunction bootstrapFunction, FlinkStateBinder stateBinder) {
try (SetContextClassLoader ignored = new SetContextClassLoader(bootstrapFunction)) {
- PersistedStates.findAndBind(functionType, bootstrapFunction, stateBinder);
+ PersistedStates.findAndBind(bootstrapFunction, stateBinder);
return bootstrapFunction;
}
}
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 b806df1..bce8364 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
@@ -21,8 +21,6 @@ package org.apache.flink.statefun.sdk.state;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import javax.annotation.Nullable;
-import org.apache.flink.statefun.sdk.FunctionType;
import org.apache.flink.statefun.sdk.StatefulFunction;
import org.apache.flink.statefun.sdk.annotations.ForRuntime;
import org.apache.flink.statefun.sdk.annotations.Persisted;
@@ -44,12 +42,6 @@ public final class PersistedStateRegistry {
private StateBinder stateBinder;
- /**
- * The type of the function that this registry is bound to. This is {@code NULL} if this registry
- * is not bounded.
- */
- @Nullable private FunctionType functionType;
-
public PersistedStateRegistry() {
this.stateBinder = new NonFaultTolerantStateBinder();
}
@@ -96,23 +88,24 @@ public final class PersistedStateRegistry {
* will also be bound to the system.
*
* @param stateBinder the new fault-tolerant state binder to use.
- * @param functionType the type of the function that this registry is being bound to.
* @throws IllegalStateException if the registry was attempted to be bound more than once.
*/
@ForRuntime
- void bind(StateBinder stateBinder, FunctionType functionType) {
- if (this.functionType != null) {
+ void bind(StateBinder stateBinder) {
+ if (isBound()) {
throw new IllegalStateException(
- "This registry was already bound to function type: "
- + this.functionType
- + ", attempting to rebind to function type: "
- + functionType);
+ "This registry was already bound to state binder: "
+ + this.stateBinder.getClass().getName()
+ + ", attempting to rebind to state binder: "
+ + stateBinder.getClass().getName());
}
this.stateBinder = Objects.requireNonNull(stateBinder);
- this.functionType = Objects.requireNonNull(functionType);
+ registeredStates.values().forEach(stateBinder::bind);
+ }
- registeredStates.values().forEach(state -> stateBinder.bind(state, functionType));
+ private boolean isBound() {
+ return stateBinder != null && !(stateBinder instanceof NonFaultTolerantStateBinder);
}
private void acceptRegistrationOrThrowIfPresent(String stateName, Object newStateObject) {
@@ -125,18 +118,17 @@ public final class PersistedStateRegistry {
}
registeredStates.put(stateName, newStateObject);
- stateBinder.bind(newStateObject, functionType);
+ stateBinder.bind(newStateObject);
}
private static final class NonFaultTolerantStateBinder extends StateBinder {
@Override
- public void bindValue(PersistedValue<?> persistedValue, FunctionType functionType) {}
+ public void bindValue(PersistedValue<?> persistedValue) {}
@Override
- public void bindTable(PersistedTable<?, ?> persistedTable, FunctionType functionType) {}
+ public void bindTable(PersistedTable<?, ?> persistedTable) {}
@Override
- public void bindAppendingBuffer(
- PersistedAppendingBuffer<?> persistedAppendingBuffer, FunctionType functionType) {}
+ public void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer) {}
}
}
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 1cfcb3e..01560cc 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,23 +18,20 @@
package org.apache.flink.statefun.sdk.state;
-import org.apache.flink.statefun.sdk.FunctionType;
-
public abstract class StateBinder {
- public abstract void bindValue(PersistedValue<?> persistedValue, FunctionType functionType);
+ public abstract void bindValue(PersistedValue<?> persistedValue);
- public abstract void bindTable(PersistedTable<?, ?> persistedTable, FunctionType functionType);
+ public abstract void bindTable(PersistedTable<?, ?> persistedTable);
- public abstract void bindAppendingBuffer(
- PersistedAppendingBuffer<?> persistedAppendingBuffer, FunctionType functionType);
+ public abstract void bindAppendingBuffer(PersistedAppendingBuffer<?> persistedAppendingBuffer);
- public final void bind(Object stateObject, FunctionType functionType) {
+ public final void bind(Object stateObject) {
if (stateObject instanceof PersistedValue) {
- bindValue((PersistedValue<?>) stateObject, functionType);
+ bindValue((PersistedValue<?>) stateObject);
} else if (stateObject instanceof PersistedTable) {
- bindTable((PersistedTable<?, ?>) stateObject, functionType);
+ bindTable((PersistedTable<?, ?>) stateObject);
} else if (stateObject instanceof PersistedAppendingBuffer) {
- bindAppendingBuffer((PersistedAppendingBuffer<?>) stateObject, functionType);
+ bindAppendingBuffer((PersistedAppendingBuffer<?>) stateObject);
} else {
throw new IllegalArgumentException("Unknown persisted state object " + stateObject);
}
[flink-statefun] 04/04: [hotfix] [core] Rename
PersistedStates.findReflectivelyAndBind
Posted by tz...@apache.org.
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 822c6e6bba716f658eec3ccd8612997aa3fd6108
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Aug 31 16:58:23 2020 +0800
[hotfix] [core] Rename PersistedStates.findReflectivelyAndBind
The renaming provides a better description that the method reflectively
searches for fields marked with the @Persisted annotation and binds it
to Flink state via a state binder.
---
.../core/functions/StatefulFunctionRepository.java | 2 +-
.../statefun/flink/core/state/PersistedStates.java | 3 ++-
.../statefun/flink/core/state/PersistedStatesTest.java | 18 +++++++++---------
.../operator/StateBootstrapFunctionRegistry.java | 2 +-
4 files changed, 13 insertions(+), 12 deletions(-)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java
index af512d0..7c1af51 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/StatefulFunctionRepository.java
@@ -64,7 +64,7 @@ final class StatefulFunctionRepository implements FunctionRepository {
functionLoader.load(functionType);
try (SetContextClassLoader ignored = new SetContextClassLoader(statefulFunction)) {
FlinkStateBinder stateBinderForType = new FlinkStateBinder(flinkState, functionType);
- PersistedStates.findAndBind(statefulFunction, stateBinderForType);
+ PersistedStates.findReflectivelyAndBind(statefulFunction, stateBinderForType);
FunctionTypeMetrics metrics = metricsFactory.forType(functionType);
return new StatefulFunction(statefulFunction, metrics, messageFactory);
}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java
index 968e4b5..3e52adf 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/state/PersistedStates.java
@@ -35,7 +35,8 @@ import org.apache.flink.statefun.sdk.state.PersistedValue;
public final class PersistedStates {
- public static void findAndBind(@Nullable Object instance, FlinkStateBinder stateBinder) {
+ public static void findReflectivelyAndBind(
+ @Nullable Object instance, FlinkStateBinder stateBinder) {
List<?> states = findReflectively(instance);
for (Object persisted : states) {
if (persisted instanceof PersistedStateRegistry) {
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
index adddcf5..ad07d5a 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/state/PersistedStatesTest.java
@@ -52,45 +52,45 @@ public class PersistedStatesTest {
@Test
public void exampleUsage() {
- PersistedStates.findAndBind(new SanityClass(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new SanityClass(), binderUnderTest);
assertThat(state.boundNames, hasItems("name", "last"));
}
@Test(expected = IllegalStateException.class)
public void nullValueField() {
- PersistedStates.findAndBind(new NullValueClass(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new NullValueClass(), binderUnderTest);
}
@Test
public void nonAnnotatedClass() {
- PersistedStates.findAndBind(new IgnoreNonAnnotated(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new IgnoreNonAnnotated(), binderUnderTest);
assertTrue(state.boundNames.isEmpty());
}
@Test
public void extendedClass() {
- PersistedStates.findAndBind(new ChildClass(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new ChildClass(), binderUnderTest);
assertThat(state.boundNames, hasItems("parent", "child"));
}
@Test(expected = IllegalArgumentException.class)
public void staticPersistedFieldsAreNotAllowed() {
- PersistedStates.findAndBind(new StaticPersistedValue(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new StaticPersistedValue(), binderUnderTest);
}
@Test
public void bindPersistedTable() {
- PersistedStates.findAndBind(new PersistedTableValue(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new PersistedTableValue(), binderUnderTest);
assertThat(state.boundNames, hasItems("table"));
}
@Test
public void bindPersistedAppendingBuffer() {
- PersistedStates.findAndBind(new PersistedAppendingBufferState(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new PersistedAppendingBufferState(), binderUnderTest);
assertThat(state.boundNames, hasItems("buffer"));
}
@@ -98,7 +98,7 @@ public class PersistedStatesTest {
@Test
public void bindDynamicState() {
DynamicState dynamicState = new DynamicState();
- PersistedStates.findAndBind(dynamicState, binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(dynamicState, binderUnderTest);
dynamicState.process();
@@ -115,7 +115,7 @@ public class PersistedStatesTest {
@Test
public void bindComposedState() {
- PersistedStates.findAndBind(new OuterClass(), binderUnderTest);
+ PersistedStates.findReflectivelyAndBind(new OuterClass(), binderUnderTest);
assertThat(state.boundNames, hasItems("inner"));
}
diff --git a/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java b/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java
index 2825f75..f99b88d 100644
--- a/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java
+++ b/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/StateBootstrapFunctionRegistry.java
@@ -117,7 +117,7 @@ public final class StateBootstrapFunctionRegistry implements Serializable {
private static StateBootstrapFunction bindState(
StateBootstrapFunction bootstrapFunction, FlinkStateBinder stateBinder) {
try (SetContextClassLoader ignored = new SetContextClassLoader(bootstrapFunction)) {
- PersistedStates.findAndBind(bootstrapFunction, stateBinder);
+ PersistedStates.findReflectivelyAndBind(bootstrapFunction, stateBinder);
return bootstrapFunction;
}
}