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-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 4f99c5f2fd38ae..fd283a4c7b44f7 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<Boolean> values, Collector<Boolean> 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<UUID, String> mapper =
+                new MapFunction<UUID, String>() {
+                    @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 {