From ccfe4317ee8264e4c82a4b19925e91fd99c9dff1 Mon Sep 17 00:00:00 2001 From: dylanhz Date: Tue, 8 Sep 2026 11:54:57 +0800 Subject: [PATCH 1/2] [FLINK-40597][core] Fix input type validation for Types.UUID Generated-by: Codex 0.147.0 (model: gpt-6) --- .../api/java/typeutils/TypeExtractor.java | 13 +++++++--- .../flink/streaming/api/DataStreamTest.java | 26 +++++++++++++++++++ 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java index 76c7e6ce39253..c05813731445b 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java @@ -77,6 +77,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.checkAndExtractLambda; import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.getAllDeclaredMethods; @@ -1498,10 +1499,16 @@ private static void validateInfo( // check for Java Basic Types if (typeInfo instanceof BasicTypeInfo) { - TypeInformation actual; // check if basic type at all - if (!(type instanceof Class) - || (actual = BasicTypeInfo.getInfoFor((Class) type)) == null) { + if (!(type instanceof Class)) { + throw new InvalidTypesException("Basic type expected."); + } + // UUID is not registered for automatic extraction to preserve existing serializers. + final TypeInformation actual = + type == UUID.class + ? BasicTypeInfo.UUID_TYPE_INFO + : BasicTypeInfo.getInfoFor((Class) type); + if (actual == null) { throw new InvalidTypesException("Basic type expected."); } // check if correct basic type diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java index 747cec98067bb..a3cebd1c66bb3 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java @@ -90,6 +90,7 @@ import java.lang.reflect.Method; import java.time.Duration; import java.util.List; +import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -836,6 +837,31 @@ public Long reduce(Long value1, Long value2) throws Exception { .isEqualTo(preferredResource7); } + @Test + void testMapWithExplicitUuidType() { + final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + // A lambda would bypass the input validation exercised by this test. + final MapFunction mapper = + new MapFunction() { + @Override + public String map(UUID value) { + return value.toString(); + } + }; + + final DataStream stream = env.fromData(Types.UUID, new UUID(0L, 0L)).map(mapper); + + assertThat(stream.getType()).isEqualTo(Types.STRING); + } + + @Test + void testUuidIsNotAutomaticallyExtracted() { + final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + + assertThat(env.fromData(new UUID(0L, 0L)).getType()) + .isEqualTo(new GenericTypeInfo<>(UUID.class)); + } + @Test void testTypeInfo() { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); From 792f7a81f696434ecbe851ffab1cf940503e5709 Mon Sep 17 00:00:00 2001 From: dylanhz Date: Mon, 14 Sep 2026 15:34:55 +0800 Subject: [PATCH 2/2] [FLINK-40597][core] Move UUID type extraction tests to TypeExtractorTest Generated-by: Codex 0.147.0 (model: gpt-6) --- .../api/java/typeutils/TypeExtractorTest.java | 21 +++++++++++++++ .../flink/streaming/api/DataStreamTest.java | 26 ------------------- 2 files changed, 21 insertions(+), 26 deletions(-) diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java index 4f99c5f2fd38a..fd283a4c7b44f 100644 --- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java +++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java @@ -77,6 +77,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -115,6 +116,26 @@ public void reduce(Iterable values, Collector out) assertThat(TypeExtractor.getForObject(true)).isEqualTo(BasicTypeInfo.BOOLEAN_TYPE_INFO); } + @Test + void testMapWithExplicitUuidType() { + // A lambda would bypass the input validation exercised by this test. + final MapFunction mapper = + new MapFunction() { + @Override + public String map(UUID value) { + return value.toString(); + } + }; + + assertThat(TypeExtractor.getMapReturnTypes(mapper, Types.UUID)).isEqualTo(Types.STRING); + } + + @Test + void testUuidIsNotAutomaticallyExtracted() { + assertThat(TypeExtractor.getForObject(new UUID(0L, 0L))) + .isEqualTo(new GenericTypeInfo<>(UUID.class)); + } + @SuppressWarnings({"unchecked", "rawtypes"}) @Test void testTupleWithBasicTypes() throws Exception { diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java index a3cebd1c66bb3..747cec98067bb 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java @@ -90,7 +90,6 @@ import java.lang.reflect.Method; import java.time.Duration; import java.util.List; -import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -837,31 +836,6 @@ public Long reduce(Long value1, Long value2) throws Exception { .isEqualTo(preferredResource7); } - @Test - void testMapWithExplicitUuidType() { - final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - // A lambda would bypass the input validation exercised by this test. - final MapFunction mapper = - new MapFunction() { - @Override - public String map(UUID value) { - return value.toString(); - } - }; - - final DataStream stream = env.fromData(Types.UUID, new UUID(0L, 0L)).map(mapper); - - assertThat(stream.getType()).isEqualTo(Types.STRING); - } - - @Test - void testUuidIsNotAutomaticallyExtracted() { - final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); - - assertThat(env.fromData(new UUID(0L, 0L)).getType()) - .isEqualTo(new GenericTypeInfo<>(UUID.class)); - } - @Test void testTypeInfo() { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();