This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c8453bb52e7 [FLINK-40796][state-processor-api] Reject key types that
cannot be hashed reliably
c8453bb52e7 is described below
commit c8453bb52e7f46e4453f93ca7ee0f7a34a075215
Author: SEPURI-SAI-KRISHNA <[email protected]>
AuthorDate: Thu Sep 24 22:51:37 2026 +0530
[FLINK-40796][state-processor-api] Reject key types that cannot be hashed
reliably
Generated-by: Claude Code (Opus 5)
---
.../state/api/StateBootstrapTransformation.java | 4 +
.../api/StateBootstrapTransformationTest.java | 60 +++++++++++++
.../streaming/api/datastream/KeyedStream.java | 98 +---------------------
.../flink/streaming/util/keys/KeySelectorUtil.java | 96 +++++++++++++++++++++
4 files changed, 162 insertions(+), 96 deletions(-)
diff --git
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
index e5f9b853ff6..1302e7e86e5 100644
---
a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
+++
b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java
@@ -43,6 +43,7 @@ import
org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import
org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner;
+import org.apache.flink.streaming.util.keys.KeySelectorUtil;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -185,6 +186,9 @@ public class StateBootstrapTransformation<T> {
// the nested bootstrapped operator needs.
DataStream<T> input = stream;
if (keySelector != null) {
+ // keyBy() used to reject key types that cannot be hashed
reliably; keep that check
+ // here now that the partitioning is built directly.
+ KeySelectorUtil.validateKeyType(keyType);
input =
new DataStream<>(
stream.getExecutionEnvironment(),
diff --git
a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateBootstrapTransformationTest.java
b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateBootstrapTransformationTest.java
index 3935a493b85..ea635db264d 100644
---
a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateBootstrapTransformationTest.java
+++
b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateBootstrapTransformationTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.state.api;
+import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
@@ -36,6 +37,8 @@ import org.apache.flink.test.util.AbstractTestBase;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for bootstrap transformations. */
class StateBootstrapTransformationTest extends AbstractTestBase {
@@ -164,6 +167,48 @@ class StateBootstrapTransformationTest extends
AbstractTestBase {
.isEqualTo(CustomKeySelector.class);
}
+ @Test
+ void testUnhashableKeyTypeIsRejected() {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<byte[]> input = env.fromData(new byte[] {1});
+
+ StateBootstrapTransformation<byte[]> transformation =
+ OperatorTransformation.bootstrapWith(input)
+ .keyBy(new ArrayKeySelector())
+ .transform(new
ExampleArrayKeyedStateBootstrapFunction());
+
+ assertThatThrownBy(
+ () ->
+ transformation.writeOperatorSubtaskStates(
+ OperatorIdentifier.forUid("uid"),
+ new HashMapStateBackend(),
+ new Path(),
+ transformation.getMaxParallelism(4)))
+ .as("An array key cannot be hashed reliably and must be
rejected")
+ .isInstanceOf(InvalidProgramException.class)
+ .hasMessageContaining("cannot be used as key");
+ }
+
+ @Test
+ void testHashableKeyTypeIsAccepted() {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<String> input = env.fromData("");
+
+ StateBootstrapTransformation<String> transformation =
+ OperatorTransformation.bootstrapWith(input)
+ .keyBy(new CustomKeySelector())
+ .transform(new ExampleKeyedStateBootstrapFunction());
+
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ transformation.writeOperatorSubtaskStates(
+ OperatorIdentifier.forUid("uid"),
+ new HashMapStateBackend(),
+ new Path(),
+ transformation.getMaxParallelism(4)));
+ }
+
private static class CustomKeySelector implements KeySelector<String,
String> {
@Override
@@ -197,4 +242,19 @@ class StateBootstrapTransformationTest extends
AbstractTestBase {
@Override
public void processElement(String value, Context ctx) throws Exception
{}
}
+
+ private static class ArrayKeySelector implements KeySelector<byte[],
byte[]> {
+
+ @Override
+ public byte[] getKey(byte[] value) throws Exception {
+ return value;
+ }
+ }
+
+ private static class ExampleArrayKeyedStateBootstrapFunction
+ extends KeyedStateBootstrapFunction<byte[], byte[]> {
+
+ @Override
+ public void processElement(byte[] value, Context ctx) throws Exception
{}
+ }
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
index e99f390276e..b49a74c6c28 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
@@ -21,19 +21,12 @@ import org.apache.flink.annotation.Experimental;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
-import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
-import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.typeutils.EnumTypeInfo;
-import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
-import org.apache.flink.api.java.typeutils.PojoTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import
org.apache.flink.runtime.asyncprocessing.operators.AsyncKeyedProcessOperator;
import org.apache.flink.runtime.asyncprocessing.operators.AsyncStreamFlatMap;
@@ -64,16 +57,12 @@ import
org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import
org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
+import org.apache.flink.streaming.util.keys.KeySelectorUtil;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.Utils;
-import org.apache.commons.lang3.StringUtils;
-
import java.time.Duration;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Stack;
import java.util.UUID;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -160,90 +149,7 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
super(stream.getExecutionEnvironment(), partitionTransformation);
this.keySelector = clean(keySelector);
- this.keyType = validateKeyType(keyType);
- }
-
- /**
- * Validates that a given type of element (as encoded by the provided
{@link TypeInformation})
- * can be used as a key in the {@code DataStream.keyBy()} operation. This
is done by searching
- * depth-first the key type and checking if each of the composite types
satisfies the required
- * conditions (see {@link #validateKeyTypeIsHashable(TypeInformation)}).
- *
- * @param keyType The {@link TypeInformation} of the key.
- */
- @SuppressWarnings("rawtypes")
- private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType)
{
- Stack<TypeInformation<?>> stack = new Stack<>();
- stack.push(keyType);
-
- List<TypeInformation<?>> unsupportedTypes = new ArrayList<>();
-
- while (!stack.isEmpty()) {
- TypeInformation<?> typeInfo = stack.pop();
-
- if (!validateKeyTypeIsHashable(typeInfo)) {
- unsupportedTypes.add(typeInfo);
- }
-
- if (typeInfo instanceof TupleTypeInfoBase) {
- for (int i = 0; i < typeInfo.getArity(); i++) {
- stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
- }
- }
- }
-
- if (!unsupportedTypes.isEmpty()) {
- throw new InvalidProgramException(
- "Type "
- + keyType
- + " cannot be used as key. Contained "
- + "UNSUPPORTED key types: "
- + StringUtils.join(unsupportedTypes, ", ")
- + ". Look "
- + "at the keyBy() documentation for the conditions
a type has to satisfy in order to be "
- + "eligible for a key.");
- }
-
- return keyType;
- }
-
- /**
- * Validates that a given type of element (as encoded by the provided
{@link TypeInformation})
- * can be used as a key in the {@code DataStream.keyBy()} operation.
- *
- * @param type The {@link TypeInformation} of the type to check.
- * @return {@code false} if:
- * <ol>
- * <li>it is a POJO type but does not override the {@link
#hashCode()} method and relies
- * on the {@link Object#hashCode()} implementation.
- * <li>it is an array of any type (see {@link
PrimitiveArrayTypeInfo}, {@link
- * BasicArrayTypeInfo}, {@link ObjectArrayTypeInfo}).
- * <li>it is enum type
- * </ol>
- * , {@code true} otherwise.
- */
- private boolean validateKeyTypeIsHashable(TypeInformation<?> type) {
- try {
- return (type instanceof PojoTypeInfo)
- ? !type.getTypeClass()
- .getMethod("hashCode")
- .getDeclaringClass()
- .equals(Object.class)
- : !(isArrayType(type) || isEnumType(type));
- } catch (NoSuchMethodException ignored) {
- // this should never happen as we are just searching for the
hashCode() method.
- }
- return false;
- }
-
- private static boolean isArrayType(TypeInformation<?> type) {
- return type instanceof PrimitiveArrayTypeInfo
- || type instanceof BasicArrayTypeInfo
- || type instanceof ObjectArrayTypeInfo;
- }
-
- private static boolean isEnumType(TypeInformation<?> type) {
- return type instanceof EnumTypeInfo;
+ this.keyType = KeySelectorUtil.validateKeyType(keyType);
}
// ------------------------------------------------------------------------
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
index 2fda4ade164..9aeb23e41f3 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.util.keys;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.common.operators.Keys;
@@ -29,11 +30,20 @@ import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.EnumTypeInfo;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
+
+import org.apache.commons.lang3.StringUtils;
import java.lang.reflect.Array;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
+import java.util.Stack;
import static java.util.Objects.requireNonNull;
@@ -117,6 +127,92 @@ public final class KeySelectorUtil {
return new OneKeySelector<>(comparator);
}
+ /**
+ * Validates that a given type of element (as encoded by the provided
{@link TypeInformation})
+ * can be used as a key. This is done by searching depth-first the key
type and checking if each
+ * of the composite types satisfies the required conditions (see {@link
+ * #isKeyTypeHashable(TypeInformation)}).
+ *
+ * @param keyType The {@link TypeInformation} of the key.
+ * @return The validated key type.
+ * @throws InvalidProgramException If the key type, or one of its
contained types, cannot be
+ * used as a key.
+ */
+ @SuppressWarnings("rawtypes")
+ public static <KEY> TypeInformation<KEY>
validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ List<TypeInformation<?>> unsupportedTypes = new ArrayList<>();
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!isKeyTypeHashable(typeInfo)) {
+ unsupportedTypes.add(typeInfo);
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+
+ if (!unsupportedTypes.isEmpty()) {
+ throw new InvalidProgramException(
+ "Type "
+ + keyType
+ + " cannot be used as key. Contained "
+ + "UNSUPPORTED key types: "
+ + StringUtils.join(unsupportedTypes, ", ")
+ + ". Look "
+ + "at the keyBy() documentation for the conditions
a type has to satisfy in order to be "
+ + "eligible for a key.");
+ }
+
+ return keyType;
+ }
+
+ /**
+ * Checks that a given type of element (as encoded by the provided {@link
TypeInformation}) can
+ * be used as a key.
+ *
+ * @param type The {@link TypeInformation} of the type to check.
+ * @return {@code false} if:
+ * <ol>
+ * <li>it is a POJO type but does not override the {@link
Object#hashCode()} method and
+ * relies on the {@link Object#hashCode()} implementation.
+ * <li>it is an array of any type (see {@link
PrimitiveArrayTypeInfo}, {@link
+ * BasicArrayTypeInfo}, {@link ObjectArrayTypeInfo}).
+ * <li>it is enum type
+ * </ol>
+ * , {@code true} otherwise.
+ */
+ private static boolean isKeyTypeHashable(TypeInformation<?> type) {
+ try {
+ return (type instanceof PojoTypeInfo)
+ ? !type.getTypeClass()
+ .getMethod("hashCode")
+ .getDeclaringClass()
+ .equals(Object.class)
+ : !(isArrayType(type) || isEnumType(type));
+ } catch (NoSuchMethodException ignored) {
+ // this should never happen as we are just searching for the
hashCode() method.
+ }
+ return false;
+ }
+
+ private static boolean isArrayType(TypeInformation<?> type) {
+ return type instanceof PrimitiveArrayTypeInfo
+ || type instanceof BasicArrayTypeInfo
+ || type instanceof ObjectArrayTypeInfo;
+ }
+
+ private static boolean isEnumType(TypeInformation<?> type) {
+ return type instanceof EnumTypeInfo;
+ }
+
// ------------------------------------------------------------------------
/** Private constructor to prevent instantiation. */
