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 76c7e6ce39253b..c05813731445b2 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 747cec98067bb9..a3cebd1c66bb3a 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<UUID, String> mapper =
+                new MapFunction<UUID, String>() {
+                    @Override
+                    public String map(UUID value) {
+                        return value.toString();
+                    }
+                };
+
+        final DataStream<String> 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();