MINOR: Add UUID type to Kafka API code generation (#7291)

Reviewers: Colin P. McCabe <cmccabe@apache.org>

.gitignore

@@ -53,3 +53,4 @@ kafkatest.egg-info/
 systest/
 *.swp
 clients/src/generated
+clients/src/generated-test

build.gradle

@@ -1017,6 +1017,14 @@ project(':clients') {
     outputs.dir("src/generated/java/org/apache/kafka/common/message")
   }
 
+  task processTestMessages(type:JavaExec) {
+    main = "org.apache.kafka.message.MessageGenerator"
+    classpath = project(':generator').sourceSets.main.runtimeClasspath
+    args = [ "src/generated-test/java/org/apache/kafka/common/message", "src/test/resources/common/message" ]
+    inputs.dir("src/test/resources/common/message")
+    outputs.dir("src/generated-test/java/org/apache/kafka/common/message")
+  }
+
   sourceSets {
     main {
       java {
@@ -1025,13 +1033,15 @@ project(':clients') {
     }
     test {
       java {
-        srcDirs = ["src/generated/java", "src/test/java"]
+        srcDirs = ["src/generated/java", "src/generated-test/java", "src/test/java"]
       }
     }
   }
 
   compileJava.dependsOn 'processMessages'
 
+  compileTestJava.dependsOn 'processTestMessages'
+
   javadoc {
     include "**/org/apache/kafka/clients/admin/*"
     include "**/org/apache/kafka/clients/consumer/*"

clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java

@@ -20,8 +20,11 @@ package org.apache.kafka.common.protocol;
 import org.apache.kafka.common.utils.Utils;
 
 import java.util.Iterator;
+import java.util.UUID;
 
 public final class MessageUtil {
+    public static final UUID ZERO_UUID = new UUID(0L, 0L);
+
     /**
      * Get the length of the UTF8 representation of a string, without allocating
      * a byte buffer for the string.

clients/src/main/java/org/apache/kafka/common/protocol/Readable.java

@@ -18,6 +18,7 @@
 package org.apache.kafka.common.protocol;
 
 import java.nio.charset.StandardCharsets;
+import java.util.UUID;
 
 public interface Readable {
     byte readByte();
@@ -54,4 +55,11 @@ public interface Readable {
         readArray(arr);
         return arr;
     }
+
+    /**
+     * Read a UUID with the most significant digits first.
+     */
+    default UUID readUUID() {
+        return new UUID(readLong(), readLong());
+    }
 }

clients/src/main/java/org/apache/kafka/common/protocol/Writable.java

@@ -18,6 +18,7 @@
 package org.apache.kafka.common.protocol;
 
 import java.nio.charset.StandardCharsets;
+import java.util.UUID;
 
 public interface Writable {
     void writeByte(byte val);
@@ -68,4 +69,12 @@ public interface Writable {
         writeShort((short) arr.length);
         writeArray(arr);
     }
+
+    /**
+     * Write a UUID with the most significant digits first.
+     */
+    default void writeUUID(UUID uuid) {
+        writeLong(uuid.getMostSignificantBits());
+        writeLong(uuid.getLeastSignificantBits());
+    }
 }

clients/src/main/java/org/apache/kafka/common/protocol/types/Field.java

@@ -75,6 +75,16 @@ public class Field {
         }
     }
 
+    public static class UUID extends Field {
+        public UUID(String name, String docString) {
+            super(name, Type.UUID, docString, false, null);
+        }
+
+        public UUID(String name, String docString, UUID defaultValue) {
+            super(name, Type.UUID, docString, true, defaultValue);
+        }
+    }
+
     public static class Int16 extends Field {
         public Int16(String name, String docString) {
             super(name, Type.INT16, docString, false, null);

clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java

@@ -21,6 +21,7 @@ import org.apache.kafka.common.record.BaseRecords;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Objects;
+import java.util.UUID;
 
 /**
  * A record that can be serialized and deserialized according to a pre-defined schema
@@ -88,6 +89,10 @@ public class Struct {
         return getLong(field.name);
     }
 
+    public UUID get(Field.UUID field) {
+        return getUUID(field.name);
+    }
+
     public Short get(Field.Int16 field) {
         return getShort(field.name);
     }
@@ -118,6 +123,12 @@ public class Struct {
         return alternative;
     }
 
+    public UUID getOrElse(Field.UUID field, UUID alternative) {
+        if (hasField(field.name))
+            return getUUID(field.name);
+        return alternative;
+    }
+
     public Short getOrElse(Field.Int16 field, short alternative) {
         if (hasField(field.name))
             return getShort(field.name);
@@ -245,6 +256,14 @@ public class Struct {
         return (Long) get(name);
     }
 
+    public UUID getUUID(BoundField field) {
+        return (UUID) get(field);
+    }
+
+    public UUID getUUID(String name) {
+        return (UUID) get(name);
+    }
+
     public Object[] getArray(BoundField field) {
         return (Object[]) get(field);
     }
@@ -342,6 +361,10 @@ public class Struct {
         return set(def.name, value);
     }
 
+    public Struct set(Field.UUID def, UUID value) {
+        return set(def.name, value);
+    }
+
     public Struct set(Field.Int16 def, short value) {
         return set(def.name, value);
     }

clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java

@@ -23,6 +23,7 @@ import org.apache.kafka.common.utils.ByteUtils;
 import org.apache.kafka.common.utils.Utils;
 
 import java.nio.ByteBuffer;
+import java.util.UUID;
 
 /**
  * A serializable type
@@ -313,6 +314,44 @@ public abstract class Type {
         }
     };
 
+    public static final DocumentedType UUID = new DocumentedType() {
+        @Override
+        public void write(ByteBuffer buffer, Object o) {
+            final java.util.UUID uuid = (java.util.UUID) o;
+            buffer.putLong(uuid.getMostSignificantBits());
+            buffer.putLong(uuid.getLeastSignificantBits());
+        }
+
+        @Override
+        public Object read(ByteBuffer buffer) {
+            return new java.util.UUID(buffer.getLong(), buffer.getLong());
+        }
+
+        @Override
+        public int sizeOf(Object o) {
+            return 16;
+        }
+
+        @Override
+        public String typeName() {
+            return "UUID";
+        }
+
+        @Override
+        public UUID validate(Object item) {
+            if (item instanceof UUID)
+                return (UUID) item;
+            else
+                throw new SchemaException(item + " is not a UUID.");
+        }
+
+        @Override
+        public String documentation() {
+            return "Represents a java.util.UUID. " +
+                    "The values are encoded using sixteen bytes in network byte order (big-endian).";
+        }
+    };
+
     public static final DocumentedType STRING = new DocumentedType() {
         @Override
         public void write(ByteBuffer buffer, Object o) {

clients/src/test/java/org/apache/kafka/common/message/TestUUIDDataTest.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.message;
+
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
+public class TestUUIDDataTest {
+
+    @Test
+    public void shouldStoreField() {
+        final UUID uuid = UUID.randomUUID();
+        final TestUUIDData out = new TestUUIDData();
+        out.setProcessId(uuid);
+        Assert.assertEquals(uuid, out.processId());
+    }
+
+    @Test
+    public void shouldDefaultField() {
+        final TestUUIDData out = new TestUUIDData();
+        Assert.assertEquals(UUID.fromString("00000000-0000-0000-0000-000000000000"), out.processId());
+    }
+
+    @Test
+    public void shouldRoundTripFieldThroughStruct() {
+        final UUID uuid = UUID.randomUUID();
+        final TestUUIDData out = new TestUUIDData();
+        out.setProcessId(uuid);
+
+        final Struct struct = out.toStruct((short) 1);
+        final TestUUIDData in = new TestUUIDData();
+        in.fromStruct(struct, (short) 1);
+
+        Assert.assertEquals(uuid, in.processId());
+    }
+
+    @Test
+    public void shouldRoundTripFieldThroughBuffer() {
+        final UUID uuid = UUID.randomUUID();
+        final TestUUIDData out = new TestUUIDData();
+        out.setProcessId(uuid);
+
+        final ByteBuffer buffer = ByteBuffer.allocate(out.size((short) 1));
+        out.write(new ByteBufferAccessor(buffer), (short) 1);
+        buffer.rewind();
+
+        final TestUUIDData in = new TestUUIDData();
+        in.read(new ByteBufferAccessor(buffer), (short) 1);
+
+        Assert.assertEquals(uuid, in.processId());
+    }
+
+    @Test
+    public void shouldImplementJVMMethods() {
+        final UUID uuid = UUID.randomUUID();
+        final TestUUIDData a = new TestUUIDData();
+        a.setProcessId(uuid);
+
+        final TestUUIDData b = new TestUUIDData();
+        b.setProcessId(uuid);
+
+        Assert.assertEquals(a, b);
+        Assert.assertEquals(a.hashCode(), b.hashCode());
+        // just tagging this on here
+        Assert.assertEquals(a.toString(), b.toString());
+    }
+}

clients/src/test/resources/common/message/TestUUID.json

@@ -0,0 +1,26 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+{
+  "name": "TestUUID",
+  "validVersions": "1",
+  "fields": [
+    {
+      "name": "processId",
+      "versions": "1",
+      "type": "uuid"
+    }
+  ],
+  "type": "header"
+}

generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java

@@ -73,6 +73,10 @@ public final class ApiMessageTypeGenerator {
         this.buffer = new CodeBuffer();
     }
 
+    public boolean hasRegisteredTypes() {
+        return !apis.isEmpty();
+    }
+
     public void registerMessageType(MessageSpec spec) {
         switch (spec.type()) {
             case REQUEST: {

generator/src/main/java/org/apache/kafka/message/FieldType.java

@@ -41,6 +41,11 @@ public interface FieldType {
         static final Int8FieldType INSTANCE = new Int8FieldType();
         private static final String NAME = "int8";
 
+        @Override
+        public boolean isInteger() {
+            return true;
+        }
+
         @Override
         public Optional<Integer> fixedLength() {
             return Optional.of(1);
@@ -56,6 +61,11 @@ public interface FieldType {
         static final Int16FieldType INSTANCE = new Int16FieldType();
         private static final String NAME = "int16";
 
+        @Override
+        public boolean isInteger() {
+            return true;
+        }
+
         @Override
         public Optional<Integer> fixedLength() {
             return Optional.of(2);
@@ -71,6 +81,11 @@ public interface FieldType {
         static final Int32FieldType INSTANCE = new Int32FieldType();
         private static final String NAME = "int32";
 
+        @Override
+        public boolean isInteger() {
+            return true;
+        }
+
         @Override
         public Optional<Integer> fixedLength() {
             return Optional.of(4);
@@ -86,6 +101,11 @@ public interface FieldType {
         static final Int64FieldType INSTANCE = new Int64FieldType();
         private static final String NAME = "int64";
 
+        @Override
+        public boolean isInteger() {
+            return true;
+        }
+
         @Override
         public Optional<Integer> fixedLength() {
             return Optional.of(8);
@@ -97,6 +117,21 @@ public interface FieldType {
         }
     }
 
+    final class UUIDFieldType implements FieldType {
+        static final UUIDFieldType INSTANCE = new UUIDFieldType();
+        private static final String NAME = "uuid";
+
+        @Override
+        public Optional<Integer> fixedLength() {
+            return Optional.of(16);
+        }
+
+        @Override
+        public String toString() {
+            return NAME;
+        }
+    }
+
     final class StringFieldType implements FieldType {
         static final StringFieldType INSTANCE = new StringFieldType();
         private static final String NAME = "string";
@@ -204,6 +239,8 @@ public interface FieldType {
                 return Int32FieldType.INSTANCE;
             case Int64FieldType.NAME:
                 return Int64FieldType.INSTANCE;
+            case UUIDFieldType.NAME:
+                return UUIDFieldType.INSTANCE;
             case StringFieldType.NAME:
                 return StringFieldType.INSTANCE;
             case BytesFieldType.NAME:
@@ -229,6 +266,10 @@ public interface FieldType {
         }
     }
 
+    default boolean isInteger() {
+        return false;
+    }
+
     /**
      * Returns true if this is an array type.
      */

generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java

@@ -22,6 +22,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Optional;
 import java.util.Set;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 /**
@@ -311,6 +312,8 @@ public final class MessageDataGenerator {
             return "int";
         } else if (field.type() instanceof FieldType.Int64FieldType) {
             return "long";
+        } else if (field.type() instanceof FieldType.UUIDFieldType) {
+            return "UUID";
         } else if (field.type().isString()) {
             return "String";
         } else if (field.type().isBytes()) {
@@ -475,6 +478,8 @@ public final class MessageDataGenerator {
             return "readable.readInt()";
         } else if (type instanceof FieldType.Int64FieldType) {
             return "readable.readLong()";
+        } else if (type instanceof FieldType.UUIDFieldType) {
+            return "readable.readUUID()";
         } else if (type.isString()) {
             return "readable.readNullableString()";
         } else if (type.isBytes()) {
@@ -581,6 +586,8 @@ public final class MessageDataGenerator {
             return "Integer";
         } else if (type instanceof FieldType.Int64FieldType) {
             return "Long";
+        } else if (type instanceof FieldType.UUIDFieldType) {
+            return "UUID";
         } else if (type.isString()) {
             return "String";
         } else if (type.isStruct()) {
@@ -601,6 +608,8 @@ public final class MessageDataGenerator {
             return String.format("struct.getInt(\"%s\")", name);
         } 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);
         } else if (type.isString()) {
             return String.format("struct.getString(\"%s\")", name);
         } else if (type.isBytes()) {
@@ -649,6 +658,8 @@ public final class MessageDataGenerator {
             return String.format("writable.writeInt(%s)", name);
         } 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);
         } else if (type instanceof FieldType.StringFieldType) {
             if (nullable) {
                 return String.format("writable.writeNullableString(%s)", name);
@@ -746,10 +757,8 @@ public final class MessageDataGenerator {
                 " cannot be nullable.");
         }
         if ((field.type() instanceof FieldType.BoolFieldType) ||
-                (field.type() instanceof FieldType.Int8FieldType) ||
+                (field.type().isInteger()) ||
-                (field.type() instanceof FieldType.Int16FieldType) ||
+                (field.type() instanceof FieldType.UUIDFieldType) ||
-                (field.type() instanceof FieldType.Int32FieldType) ||
-                (field.type() instanceof FieldType.Int64FieldType) ||
                 (field.type() instanceof FieldType.StringFieldType)) {
             boolean maybeAbsent =
                 generateVersionCheck(curVersions, field.versions());
@@ -1039,16 +1048,17 @@ public final class MessageDataGenerator {
         } else if (field.type() instanceof FieldType.Int64FieldType) {
             buffer.printf("hashCode = 31 * hashCode + ((int) (%s >> 32) ^ (int) %s);%n",
                 field.camelCaseName(), field.camelCaseName());
-        } else if (field.type().isString()) {
-            buffer.printf("hashCode = 31 * hashCode + (%s == null ? 0 : %s.hashCode());%n",
-                field.camelCaseName(), field.camelCaseName());
         } else if (field.type().isBytes()) {
             headerGenerator.addImport(MessageGenerator.ARRAYS_CLASS);
             buffer.printf("hashCode = 31 * hashCode + Arrays.hashCode(%s);%n",
                 field.camelCaseName());
-        } else if (field.type().isStruct() || field.type().isArray()) {
+        } else if (field.type().isStruct()
+                   || field.type().isArray()
+                   || field.type().isString()
+                   || field.type() instanceof  FieldType.UUIDFieldType
+        ) {
             buffer.printf("hashCode = 31 * hashCode + (%s == null ? 0 : %s.hashCode());%n",
-                field.camelCaseName(), field.camelCaseName());
+                          field.camelCaseName(), field.camelCaseName());
         } else {
             throw new RuntimeException("Unsupported field type " + field.type());
         }
@@ -1078,7 +1088,8 @@ public final class MessageDataGenerator {
         } else if ((field.type() instanceof FieldType.Int8FieldType) ||
                 (field.type() instanceof FieldType.Int16FieldType) ||
                 (field.type() instanceof FieldType.Int32FieldType) ||
-                (field.type() instanceof FieldType.Int64FieldType)) {
+                (field.type() instanceof FieldType.Int64FieldType) ||
+                (field.type() instanceof FieldType.UUIDFieldType)) {
             buffer.printf("+ \"%s%s=\" + %s%n",
                 prefix, field.camelCaseName(), field.camelCaseName());
         } else if (field.type().isString()) {
@@ -1227,6 +1238,19 @@ public final class MessageDataGenerator {
                 }
                 return field.defaultString() + "L";
             }
+        } else if (field.type() instanceof FieldType.UUIDFieldType) {
+            headerGenerator.addImport(MessageGenerator.UUID_CLASS);
+            if (field.defaultString().isEmpty()) {
+                return "org.apache.kafka.common.protocol.MessageUtil.ZERO_UUID";
+            } else {
+                try {
+                    UUID.fromString(field.defaultString());
+                } catch (IllegalArgumentException e) {
+                    throw new RuntimeException("Invalid default for uuid field " +
+                        field.name() + ": " + field.defaultString(), e);
+                }
+                return "UUID.fromString(\"" + field.defaultString() + "\")";
+            }
         } else if (field.type() instanceof FieldType.StringFieldType) {
             if (field.defaultString().equals("null")) {
                 validateNullDefault(field);

generator/src/main/java/org/apache/kafka/message/MessageGenerator.java

@@ -79,6 +79,8 @@ public final class MessageGenerator {
 
     static final String BYTES_CLASS = "org.apache.kafka.common.utils.Bytes";
 
+    static final String UUID_CLASS = "java.util.UUID";
+
     static final String REQUEST_SUFFIX = "Request";
 
     static final String RESPONSE_SUFFIX = "Response";
@@ -122,13 +124,15 @@ public final class MessageGenerator {
                 }
             }
         }
-        Path factoryOutputPath = Paths.get(outputDir, API_MESSAGE_TYPE_JAVA);
+        if (messageTypeGenerator.hasRegisteredTypes()) {
-        outputFileNames.add(API_MESSAGE_TYPE_JAVA);
+            Path factoryOutputPath = Paths.get(outputDir, API_MESSAGE_TYPE_JAVA);
-        try (BufferedWriter writer = Files.newBufferedWriter(factoryOutputPath)) {
+            outputFileNames.add(API_MESSAGE_TYPE_JAVA);
-            messageTypeGenerator.generate();
+            try (BufferedWriter writer = Files.newBufferedWriter(factoryOutputPath)) {
-            messageTypeGenerator.write(writer);
+                messageTypeGenerator.generate();
+                messageTypeGenerator.write(writer);
+            }
+            numProcessed++;
         }
-        numProcessed++;
         try (DirectoryStream<Path> directoryStream = Files.
                 newDirectoryStream(Paths.get(outputDir))) {
             for (Path outputPath : directoryStream) {

generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java

@@ -188,6 +188,12 @@ final class SchemaGenerator {
                 throw new RuntimeException("Type " + type + " cannot be nullable.");
             }
             return "Type.INT64";
+        } else if (type instanceof FieldType.UUIDFieldType) {
+            headerGenerator.addImport(MessageGenerator.TYPE_CLASS);
+            if (nullable) {
+                throw new RuntimeException("Type " + type + " cannot be nullable.");
+            }
+            return "Type.UUID";
         } else if (type instanceof FieldType.StringFieldType) {
             headerGenerator.addImport(MessageGenerator.TYPE_CLASS);
             return nullable ? "Type.NULLABLE_STRING" : "Type.STRING";