diff --git a/clients/src/main/java/org/apache/kafka/common/UUID.java b/clients/src/main/java/org/apache/kafka/common/Uuid.java
similarity index 71%
rename from clients/src/main/java/org/apache/kafka/common/UUID.java
rename to clients/src/main/java/org/apache/kafka/common/Uuid.java
index 29abdaa44c8..65f56a24abd 100644
--- a/clients/src/main/java/org/apache/kafka/common/UUID.java
+++ b/clients/src/main/java/org/apache/kafka/common/Uuid.java
@@ -22,68 +22,66 @@ import java.util.Base64;
 /**
  * This class defines an immutable universally unique identifier (UUID). It represents a 128-bit value.
  * More specifically, the random UUIDs generated by this class are variant 2 (Leach-Salz) version 4 UUIDs.
- * This is the same type of UUID as the ones generated by java.util.UUID except that the toString() method prints
+ * This is the same type of UUID as the ones generated by java.util.UUID. The toString() method prints
  * using the base64 string encoding. Likewise, the fromString method expects a base64 string encoding.
  */
-public class UUID {
+public class Uuid {
 
     private static final java.util.UUID SENTINEL_ID_INTERNAL = new java.util.UUID(0L, 1L);
 
     /**
-     * A UUID that represents a null or empty UUID. Will never be returned by the randomUUID method
+     * A UUID that represents a null or empty UUID. Will never be returned by the randomUuid method.
      */
-    public static final UUID ZERO_UUID = new UUID(new java.util.UUID(0L, 0L));
+    public static final Uuid ZERO_UUID = new Uuid(0L, 0L);
     private static final java.util.UUID ZERO_ID_INTERNAL = new java.util.UUID(0L, 0L);
 
-    private final java.util.UUID uuid;
+    private final long mostSignificantBits;
+    private final long leastSignificantBits;
 
     /**
      * Constructs a 128-bit type 4 UUID where the first long represents the the most significant 64 bits
      * and the second long represents the least significant 64 bits.
      */
-    public UUID(long mostSigBits, long leastSigBits) {
-        this.uuid = new java.util.UUID(mostSigBits, leastSigBits);
-    }
-
-    private UUID(java.util.UUID uuid) {
-        this.uuid = uuid;
+    public Uuid(long mostSigBits, long leastSigBits) {
+        this.mostSignificantBits = mostSigBits;
+        this.leastSignificantBits = leastSigBits;
     }
 
     /**
      * Static factory to retrieve a type 4 (pseudo randomly generated) UUID.
      */
-    public static UUID randomUUID() {
+    public static Uuid randomUuid() {
         java.util.UUID uuid = java.util.UUID.randomUUID();
         while (uuid.equals(SENTINEL_ID_INTERNAL) || uuid.equals(ZERO_ID_INTERNAL)) {
             uuid = java.util.UUID.randomUUID();
         }
-        return new UUID(uuid);
+        return new Uuid(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
     }
 
     /**
      * Returns the most significant bits of the UUID's 128 value.
      */
     public long getMostSignificantBits() {
-        return uuid.getMostSignificantBits();
+        return this.mostSignificantBits;
     }
 
     /**
      * Returns the least significant bits of the UUID's 128 value.
      */
     public long getLeastSignificantBits() {
-        return uuid.getLeastSignificantBits();
+        return this.leastSignificantBits;
     }
 
     /**
-     * Returns true iff obj is another UUID represented by the same two long values.
+     * Returns true iff obj is another Uuid represented by the same two long values.
      */
     @Override
     public boolean equals(Object obj) {
         if ((null == obj) || (obj.getClass() != this.getClass()))
             return false;
-        UUID id = (UUID) obj;
-        return this.getMostSignificantBits() == id.getMostSignificantBits() &&
-                this.getLeastSignificantBits() == id.getLeastSignificantBits();
+        Uuid id = (Uuid) obj;
+        return this.mostSignificantBits == id.mostSignificantBits &&
+                this.leastSignificantBits == id.leastSignificantBits;
     }
 
     /**
@@ -91,7 +89,8 @@ public class UUID {
      */
     @Override
     public int hashCode() {
-        return uuid.hashCode();
+        long xor = mostSignificantBits ^ leastSignificantBits;
+        return (int) (xor >> 32) ^ (int) xor;
     }
 
     /**
@@ -99,22 +98,22 @@ public class UUID {
      */
     @Override
     public String toString() {
-        return Base64.getUrlEncoder().withoutPadding().encodeToString(getBytesFromUuid(uuid));
+        return Base64.getUrlEncoder().withoutPadding().encodeToString(getBytesFromUuid());
     }
 
     /**
      * Creates a UUID based on a base64 string encoding used in the toString() method.
      */
-    public static UUID fromString(String str) {
+    public static Uuid fromString(String str) {
         ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(str));
-        return new UUID(uuidBytes.getLong(), uuidBytes.getLong());
+        return new Uuid(uuidBytes.getLong(), uuidBytes.getLong());
     }
 
-    private byte[] getBytesFromUuid(java.util.UUID uuid) {
+    private byte[] getBytesFromUuid() {
         // Extract bytes for uuid which is 128 bits (or 16 bytes) long.
         ByteBuffer uuidBytes = ByteBuffer.wrap(new byte[16]);
-        uuidBytes.putLong(uuid.getMostSignificantBits());
-        uuidBytes.putLong(uuid.getLeastSignificantBits());
+        uuidBytes.putLong(this.mostSignificantBits);
+        uuidBytes.putLong(this.leastSignificantBits);
         return uuidBytes.array();
     }
 }
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java b/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java
index 805b3c73166..8bf9be8883b 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java
@@ -17,7 +17,7 @@
 
 package org.apache.kafka.common.protocol;
 
-import org.apache.kafka.common.UUID;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.protocol.types.RawTaggedField;
 import org.apache.kafka.common.record.MemoryRecords;
 
@@ -67,7 +67,7 @@ public interface Readable {
     /**
      * Read a UUID with the most significant digits first.
      */
-    default UUID readUUID() {
-        return new UUID(readLong(), readLong());
+    default Uuid readUuid() {
+        return new Uuid(readLong(), readLong());
     }
 }
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Writable.java b/clients/src/main/java/org/apache/kafka/common/protocol/Writable.java
index ac665097411..1ca59d04e0f 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/Writable.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/Writable.java
@@ -17,7 +17,7 @@
 
 package org.apache.kafka.common.protocol;
 
-import org.apache.kafka.common.UUID;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.record.BaseRecords;
 import org.apache.kafka.common.record.MemoryRecords;
 
@@ -44,7 +44,7 @@ public interface Writable {
         }
     }
 
-    default void writeUUID(UUID uuid) {
+    default void writeUuid(Uuid uuid) {
         writeLong(uuid.getMostSignificantBits());
         writeLong(uuid.getLeastSignificantBits());
     }
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
index e083ba06b18..e6e4a6fdc2c 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.protocol.types;
 
-import org.apache.kafka.common.UUID;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.record.BaseRecords;
 
 import java.nio.ByteBuffer;
@@ -89,8 +89,8 @@ public class Struct {
         return getLong(field.name);
     }
 
-    public UUID get(Field.UUID field) {
-        return getUUID(field.name);
+    public Uuid get(Field.UUID field) {
+        return getUuid(field.name);
     }
 
     public Short get(Field.Int16 field) {
@@ -127,9 +127,9 @@ public class Struct {
         return alternative;
     }
 
-    public UUID getOrElse(Field.UUID field, UUID alternative) {
+    public Uuid getOrElse(Field.UUID field, Uuid alternative) {
         if (hasField(field.name))
-            return getUUID(field.name);
+            return getUuid(field.name);
         return alternative;
     }
 
@@ -266,12 +266,12 @@ public class Struct {
         return (Long) get(name);
     }
 
-    public UUID getUUID(BoundField field) {
-        return (UUID) get(field);
+    public Uuid getUuid(BoundField field) {
+        return (Uuid) get(field);
     }
 
-    public UUID getUUID(String name) {
-        return (UUID) get(name);
+    public Uuid getUuid(String name) {
+        return (Uuid) get(name);
     }
 
     public Double getDouble(BoundField field) {
@@ -379,7 +379,7 @@ public class Struct {
         return set(def.name, value);
     }
 
-    public Struct set(Field.UUID def, UUID value) {
+    public Struct set(Field.UUID def, Uuid value) {
         return set(def.name, value);
     }
 
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java
index dd7bffa56a7..43a77fc29c3 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.protocol.types;
 
-import org.apache.kafka.common.UUID;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.record.BaseRecords;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.utils.ByteUtils;
@@ -331,14 +331,14 @@ public abstract class Type {
     public static final DocumentedType UUID = new DocumentedType() {
         @Override
         public void write(ByteBuffer buffer, Object o) {
-            final UUID uuid = (UUID) o;
+            final Uuid uuid = (Uuid) o;
             buffer.putLong(uuid.getMostSignificantBits());
             buffer.putLong(uuid.getLeastSignificantBits());
         }
 
         @Override
         public Object read(ByteBuffer buffer) {
-            return new UUID(buffer.getLong(), buffer.getLong());
+            return new Uuid(buffer.getLong(), buffer.getLong());
         }
 
         @Override
@@ -352,16 +352,16 @@ public abstract class Type {
         }
 
         @Override
-        public UUID validate(Object item) {
-            if (item instanceof UUID)
-                return (UUID) item;
+        public Uuid validate(Object item) {
+            if (item instanceof Uuid)
+                return (Uuid) item;
             else
-                throw new SchemaException(item + " is not a UUID.");
+                throw new SchemaException(item + " is not a Uuid.");
         }
 
         @Override
         public String documentation() {
-            return "Represents a type 4 immutable universally unique identifier (UUID). " +
+            return "Represents a type 4 immutable universally unique identifier (Uuid). " +
                     "The values are encoded using sixteen bytes in network byte order (big-endian).";
         }
     };
diff --git a/clients/src/test/java/org/apache/kafka/common/UUIDTest.java b/clients/src/test/java/org/apache/kafka/common/UuidTest.java
similarity index 59%
rename from clients/src/test/java/org/apache/kafka/common/UUIDTest.java
rename to clients/src/test/java/org/apache/kafka/common/UuidTest.java
index d8406a5c2fd..99e85bfe687 100644
--- a/clients/src/test/java/org/apache/kafka/common/UUIDTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/UuidTest.java
@@ -21,50 +21,61 @@ import org.junit.Test;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 
-public class UUIDTest {
+public class UuidTest {
 
     @Test
     public void testSignificantBits() {
-        UUID id = new UUID(34L, 98L);
+        Uuid id = new Uuid(34L, 98L);
 
         assertEquals(id.getMostSignificantBits(), 34L);
         assertEquals(id.getLeastSignificantBits(), 98L);
     }
 
     @Test
-    public void testUUIDEquality() {
-        UUID id1 = new UUID(12L, 13L);
-        UUID id2 = new UUID(12L, 13L);
-        UUID id3 = new UUID(24L, 38L);
+    public void testUuidEquality() {
+        Uuid id1 = new Uuid(12L, 13L);
+        Uuid id2 = new Uuid(12L, 13L);
+        Uuid id3 = new Uuid(24L, 38L);
 
-        assertEquals(UUID.ZERO_UUID, UUID.ZERO_UUID);
+        assertEquals(Uuid.ZERO_UUID, Uuid.ZERO_UUID);
         assertEquals(id1, id2);
         assertNotEquals(id1, id3);
 
-        assertEquals(UUID.ZERO_UUID.hashCode(), UUID.ZERO_UUID.hashCode());
+        assertEquals(Uuid.ZERO_UUID.hashCode(), Uuid.ZERO_UUID.hashCode());
         assertEquals(id1.hashCode(), id2.hashCode());
         assertNotEquals(id1.hashCode(), id3.hashCode());
     }
+    
+    @Test
+    public void testHashCode() {
+        Uuid id1 = new Uuid(16L, 7L);
+        Uuid id2 = new Uuid(1043L, 20075L);
+        Uuid id3 = new Uuid(104312423523523L, 200732425676585L);
+        
+        assertEquals(23, id1.hashCode());
+        assertEquals(19064, id2.hashCode());
+        assertEquals(-2011255899, id3.hashCode());
+    }
 
     @Test
     public void testStringConversion() {
-        UUID id = UUID.randomUUID();
+        Uuid id = Uuid.randomUuid();
         String idString = id.toString();
 
-        assertEquals(UUID.fromString(idString), id);
+        assertEquals(Uuid.fromString(idString), id);
 
-        String zeroIdString = UUID.ZERO_UUID.toString();
+        String zeroIdString = Uuid.ZERO_UUID.toString();
 
-        assertEquals(UUID.fromString(zeroIdString), UUID.ZERO_UUID);
+        assertEquals(Uuid.fromString(zeroIdString), Uuid.ZERO_UUID);
     }
 
     @Test
-    public void testRandomUUID() {
-        UUID randomID = UUID.randomUUID();
-        // reservedSentinel is based on the value of SENTINEL_ID_INTERNAL in UUID.
-        UUID reservedSentinel = new UUID(0L, 1L);
+    public void testRandomUuid() {
+        Uuid randomID = Uuid.randomUuid();
+        // reservedSentinel is based on the value of SENTINEL_ID_INTERNAL in Uuid.
+        Uuid reservedSentinel = new Uuid(0L, 1L);
 
-        assertNotEquals(randomID, UUID.ZERO_UUID);
+        assertNotEquals(randomID, Uuid.ZERO_UUID);
         assertNotEquals(randomID, reservedSentinel);
     }
 }
diff --git a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
index adc71e3e860..f02db2374ea 100644
--- a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
@@ -20,7 +20,7 @@ package org.apache.kafka.common.message;
 import com.fasterxml.jackson.databind.JsonNode;
 
 import org.apache.kafka.common.IsolationLevel;
-import org.apache.kafka.common.UUID;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopic;
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopicCollection;
@@ -744,7 +744,7 @@ public final class MessageTest {
         ));
         message.setMyTaggedStruct(new SimpleExampleMessageData.TaggedStruct().setStructId("abc"));
 
-        message.setProcessId(UUID.randomUUID());
+        message.setProcessId(Uuid.randomUuid());
         message.setMyNullableString("notNull");
         message.setMyInt16((short) 3);
         message.setMyString("test string");
diff --git a/clients/src/test/java/org/apache/kafka/common/message/SimpleExampleMessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/SimpleExampleMessageTest.java
index c89a5c0040b..b3e45e5a64e 100644
--- a/clients/src/test/java/org/apache/kafka/common/message/SimpleExampleMessageTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/message/SimpleExampleMessageTest.java
@@ -17,7 +17,7 @@
 package org.apache.kafka.common.message;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import org.apache.kafka.common.UUID;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
 import org.apache.kafka.common.protocol.ObjectSerializationCache;
@@ -41,7 +41,7 @@ public class SimpleExampleMessageTest {
 
     @Test
     public void shouldStoreField() {
-        final UUID uuid = UUID.randomUUID();
+        final Uuid uuid = Uuid.randomUuid();
         final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
 
         final SimpleExampleMessageData out = new SimpleExampleMessageData();
@@ -61,7 +61,7 @@ public class SimpleExampleMessageTest {
     public void shouldThrowIfCannotWriteNonIgnorableField() {
         // processId is not supported in v0 and is not marked as ignorable
 
-        final SimpleExampleMessageData out = new SimpleExampleMessageData().setProcessId(UUID.randomUUID());
+        final SimpleExampleMessageData out = new SimpleExampleMessageData().setProcessId(Uuid.randomUuid());
         assertThrows(UnsupportedVersionException.class, () ->
                 out.write(new ByteBufferAccessor(ByteBuffer.allocate(64)), new ObjectSerializationCache(), (short) 0));
         assertThrows(UnsupportedVersionException.class, () -> out.toStruct((short) 0));
@@ -70,14 +70,14 @@ public class SimpleExampleMessageTest {
     @Test
     public void shouldDefaultField() {
         final SimpleExampleMessageData out = new SimpleExampleMessageData();
-        assertEquals(UUID.fromString("AAAAAAAAAAAAAAAAAAAAAA"), out.processId());
+        assertEquals(Uuid.fromString("AAAAAAAAAAAAAAAAAAAAAA"), out.processId());
         assertEquals(ByteUtils.EMPTY_BUF, out.zeroCopyByteBuffer());
         assertEquals(ByteUtils.EMPTY_BUF, out.nullableZeroCopyByteBuffer());
     }
 
     @Test
     public void shouldRoundTripFieldThroughStruct() {
-        final UUID uuid = UUID.randomUUID();
+        final Uuid uuid = Uuid.randomUuid();
         final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
         final SimpleExampleMessageData out = new SimpleExampleMessageData();
         out.setProcessId(uuid);
@@ -96,7 +96,7 @@ public class SimpleExampleMessageTest {
 
     @Test
     public void shouldRoundTripFieldThroughStructWithNullable() {
-        final UUID uuid = UUID.randomUUID();
+        final Uuid uuid = Uuid.randomUuid();
         final ByteBuffer buf1 = ByteBuffer.wrap(new byte[] {1, 2, 3});
         final ByteBuffer buf2 = ByteBuffer.wrap(new byte[] {4, 5, 6});
         final SimpleExampleMessageData out = new SimpleExampleMessageData();
@@ -118,7 +118,7 @@ public class SimpleExampleMessageTest {
 
     @Test
     public void shouldRoundTripFieldThroughBuffer() {
-        final UUID uuid = UUID.randomUUID();
+        final Uuid uuid = Uuid.randomUuid();
         final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
         final SimpleExampleMessageData out = new SimpleExampleMessageData();
         out.setProcessId(uuid);
@@ -141,7 +141,7 @@ public class SimpleExampleMessageTest {
 
     @Test
     public void shouldRoundTripFieldThroughBufferWithNullable() {
-        final UUID uuid = UUID.randomUUID();
+        final Uuid uuid = Uuid.randomUuid();
         final ByteBuffer buf1 = ByteBuffer.wrap(new byte[] {1, 2, 3});
         final ByteBuffer buf2 = ByteBuffer.wrap(new byte[] {4, 5, 6});
         final SimpleExampleMessageData out = new SimpleExampleMessageData();
@@ -167,7 +167,7 @@ public class SimpleExampleMessageTest {
 
     @Test
     public void shouldImplementEqualsAndHashCode() {
-        final UUID uuid = UUID.randomUUID();
+        final Uuid uuid = Uuid.randomUuid();
         final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
         final SimpleExampleMessageData a = new SimpleExampleMessageData();
         a.setProcessId(uuid);
@@ -259,13 +259,13 @@ public class SimpleExampleMessageTest {
     public void testTaggedUuid() {
         testRoundTrip(new SimpleExampleMessageData(),
             message -> assertEquals(
-                UUID.fromString("212d54944a8b4fdf94b388b470beb367"),
+                Uuid.fromString("212d54944a8b4fdf94b388b470beb367"),
                 message.taggedUuid()));
 
         testRoundTrip(new SimpleExampleMessageData().
-                setTaggedUuid(UUID.fromString("0123456789abcdef0123456789abcdef")),
+                setTaggedUuid(Uuid.fromString("0123456789abcdef0123456789abcdef")),
             message -> assertEquals(
-                UUID.fromString("0123456789abcdef0123456789abcdef"),
+                Uuid.fromString("0123456789abcdef0123456789abcdef"),
                 message.taggedUuid()));
     }
 
diff --git a/generator/src/main/java/org/apache/kafka/message/FieldSpec.java b/generator/src/main/java/org/apache/kafka/message/FieldSpec.java
index 201ae2e70a7..3d1c67754da 100644
--- a/generator/src/main/java/org/apache/kafka/message/FieldSpec.java
+++ b/generator/src/main/java/org/apache/kafka/message/FieldSpec.java
@@ -361,7 +361,7 @@ public final class FieldSpec {
         } else if (type instanceof FieldType.UUIDFieldType) {
             headerGenerator.addImport(MessageGenerator.UUID_CLASS);
             if (fieldDefault.isEmpty()) {
-                return "UUID.ZERO_UUID";
+                return "Uuid.ZERO_UUID";
             } else {
                 try {
                     ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(fieldDefault));
@@ -372,7 +372,7 @@ public final class FieldSpec {
                         name + ": " + fieldDefault, e);
                 }
                 headerGenerator.addImport(MessageGenerator.UUID_CLASS);
-                return "UUID.fromString(\"" + fieldDefault + "\")";
+                return "Uuid.fromString(\"" + fieldDefault + "\")";
             }
         } else if (type instanceof FieldType.Float64FieldType) {
             if (fieldDefault.isEmpty()) {
@@ -463,7 +463,7 @@ public final class FieldSpec {
             return "long";
         } else if (type instanceof FieldType.UUIDFieldType) {
             headerGenerator.addImport(MessageGenerator.UUID_CLASS);
-            return "UUID";
+            return "Uuid";
         } else if (type instanceof FieldType.Float64FieldType) {
             return "double";
         } else if (type.isString()) {
diff --git a/generator/src/main/java/org/apache/kafka/message/FieldType.java b/generator/src/main/java/org/apache/kafka/message/FieldType.java
index ee37210fa92..bb872f8d168 100644
--- a/generator/src/main/java/org/apache/kafka/message/FieldType.java
+++ b/generator/src/main/java/org/apache/kafka/message/FieldType.java
@@ -129,7 +129,7 @@ public interface FieldType {
         @Override
         public String getBoxedJavaType(HeaderGenerator headerGenerator) {
             headerGenerator.addImport(MessageGenerator.UUID_CLASS);
-            return "UUID";
+            return "Uuid";
         }
 
         @Override
diff --git a/generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java b/generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
index db265056417..94a79cb403d 100644
--- a/generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
@@ -177,7 +177,7 @@ public final class JsonConverterGenerator implements MessageClassGenerator {
             buffer.printf("}%n");
             headerGenerator.addImport(MessageGenerator.UUID_CLASS);
             buffer.printf("%s;%n", target.assignmentStatement(String.format(
-                "UUID.fromString(%s.asText())", target.sourceVariable())));
+                "Uuid.fromString(%s.asText())", target.sourceVariable())));
         } else if (target.field().type() instanceof FieldType.Float64FieldType) {
             headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
             buffer.printf("%s;%n", target.assignmentStatement(
diff --git a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
index 67e6b8226eb..97a27d92b22 100644
--- a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
@@ -559,7 +559,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
         } else if (type instanceof FieldType.Int64FieldType) {
             return "_readable.readLong()";
         } else if (type instanceof FieldType.UUIDFieldType) {
-            return "_readable.readUUID()";
+            return "_readable.readUuid()";
         } else if (type instanceof FieldType.Float64FieldType) {
             return "_readable.readDouble()";
         } else if (type.isStruct()) {
@@ -831,7 +831,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
         } else if (type instanceof FieldType.Int64FieldType) {
             return String.format("struct.getLong(\"%s\")", name);
         } else if (type instanceof FieldType.UUIDFieldType) {
-            return String.format("struct.getUUID(\"%s\")", name);
+            return String.format("struct.getUuid(\"%s\")", name);
         } else if (type instanceof FieldType.Float64FieldType) {
             return String.format("struct.getDouble(\"%s\")", name);
         } else if (type.isString()) {
@@ -1044,7 +1044,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
         } else if (type instanceof FieldType.Int64FieldType) {
             return String.format("_writable.writeLong(%s)", name);
         } else if (type instanceof FieldType.UUIDFieldType) {
-            return String.format("_writable.writeUUID(%s)", name);
+            return String.format("_writable.writeUuid(%s)", name);
         } else if (type instanceof FieldType.Float64FieldType) {
             return String.format("_writable.writeDouble(%s)", name);
         } else if (type instanceof FieldType.StructType) {
diff --git a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
index 507d53a39c7..3b357a566dc 100644
--- a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
@@ -94,7 +94,7 @@ public final class MessageGenerator {
 
     static final String BYTES_CLASS = "org.apache.kafka.common.utils.Bytes";
 
-    static final String UUID_CLASS = "org.apache.kafka.common.UUID";
+    static final String UUID_CLASS = "org.apache.kafka.common.Uuid";
 
     static final String BASE_RECORDS_CLASS = "org.apache.kafka.common.record.BaseRecords";
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java
index a939fc3b0a8..2cb7ca7f4b0 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java
@@ -21,6 +21,8 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
 import org.apache.kafka.common.protocol.ObjectSerializationCache;
 import org.apache.kafka.streams.errors.TaskAssignmentException;
@@ -87,7 +89,7 @@ public class SubscriptionInfo {
         validateVersions(version, latestSupportedVersion);
         final SubscriptionInfoData data = new SubscriptionInfoData();
         data.setVersion(version);
-        data.setProcessId(new org.apache.kafka.common.UUID(processId.getMostSignificantBits(),
+        data.setProcessId(new Uuid(processId.getMostSignificantBits(),
                 processId.getLeastSignificantBits()));
 
         if (version >= 2) {