You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2021/01/12 20:46:42 UTC
[kafka] branch trunk updated: KAFKA-12180: Implement the KIP-631
message generator changes
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 1e4d335 KAFKA-12180: Implement the KIP-631 message generator changes
1e4d335 is described below
commit 1e4d33524510769aff819ac9ef780ae1f904c3bc
Author: Colin P. Mccabe <cm...@confluent.io>
AuthorDate: Mon Jan 11 14:17:52 2021 -0800
KAFKA-12180: Implement the KIP-631 message generator changes
* Implement the uint16 type
* Implement MetadataRecordType and MetadataJsonConverters
Reviewers: Jason Gustafson <ja...@confluent.io>
---
checkstyle/suppressions.xml | 2 +-
.../apache/kafka/common/protocol/MessageUtil.java | 9 ++
.../org/apache/kafka/common/protocol/Readable.java | 4 +
.../org/apache/kafka/common/protocol/Writable.java | 6 +
.../apache/kafka/common/protocol/types/Field.java | 6 +
.../apache/kafka/common/protocol/types/Struct.java | 20 +++
.../apache/kafka/common/protocol/types/Type.java | 38 +++++
.../common/message/SimpleExampleMessageTest.java | 17 ++
.../common/message/SimpleExampleMessage.json | 3 +-
.../java/org/apache/kafka/message/FieldSpec.java | 19 +++
.../java/org/apache/kafka/message/FieldType.java | 22 +++
.../kafka/message/JsonConverterGenerator.java | 10 +-
.../apache/kafka/message/MessageDataGenerator.java | 20 +++
.../org/apache/kafka/message/MessageGenerator.java | 53 +++---
.../org/apache/kafka/message/MessageSpecType.java | 18 +++
.../message/MetadataJsonConvertersGenerator.java | 123 ++++++++++++++
.../kafka/message/MetadataRecordTypeGenerator.java | 180 +++++++++++++++++++++
.../org/apache/kafka/message/SchemaGenerator.java | 6 +
18 files changed, 533 insertions(+), 23 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 0e348d7..5871947 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -16,7 +16,7 @@
<suppress checks="JavaNCSS"
files="(ApiMessageType|FieldSpec|MessageDataGenerator).java"/>
<suppress checks="MethodLength"
- files="MessageDataGenerator.java"/>
+ files="(FieldSpec|MessageDataGenerator).java"/>
<suppress id="dontUseSystemExit"
files="MessageGenerator.java"/>
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java b/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java
index d3812bb..c3b7e37 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java
@@ -87,6 +87,15 @@ public final class MessageUtil {
return (short) value;
}
+ public static int jsonNodeToUnsignedShort(JsonNode node, String about) {
+ int value = jsonNodeToInt(node, about);
+ if (value < 0 || value > 65535) {
+ throw new RuntimeException(about + ": value " + value +
+ " does not fit in a 16-bit unsigned integer.");
+ }
+ return value;
+ }
+
public static int jsonNodeToInt(JsonNode node, String about) {
if (node.isInt()) {
return node.asInt();
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 8bf9be8..46879cd 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
@@ -70,4 +70,8 @@ public interface Readable {
default Uuid readUuid() {
return new Uuid(readLong(), readLong());
}
+
+ default int readUnsignedShort() {
+ return Short.toUnsignedInt(readShort());
+ }
}
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 1ca59d0..8dbec87 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
@@ -48,4 +48,10 @@ public interface Writable {
writeLong(uuid.getMostSignificantBits());
writeLong(uuid.getLeastSignificantBits());
}
+
+ default void writeUnsignedShort(int i) {
+ // The setter functions in the generated code prevent us from setting
+ // ints outside the valid range of a short.
+ writeShort((short) i);
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/Field.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/Field.java
index 7d5c3f5..44726f8 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/Field.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/Field.java
@@ -91,6 +91,12 @@ public class Field {
}
}
+ public static class Uint16 extends Field {
+ public Uint16(String name, String docString) {
+ super(name, Type.UINT16, docString, false, null);
+ }
+ }
+
public static class Float64 extends Field {
public Float64(String name, String docString) {
super(name, Type.FLOAT64, docString, false, null);
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 e6e4a6f..9b9b5e6 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
@@ -93,6 +93,10 @@ public class Struct {
return getUuid(field.name);
}
+ public Integer get(Field.Uint16 field) {
+ return getInt(field.name);
+ }
+
public Short get(Field.Int16 field) {
return getShort(field.name);
}
@@ -246,6 +250,14 @@ public class Struct {
return (Short) get(name);
}
+ public Integer getUnsignedShort(BoundField field) {
+ return (Integer) get(field);
+ }
+
+ public Integer getUnsignedShort(String name) {
+ return (Integer) get(name);
+ }
+
public Integer getInt(BoundField field) {
return (Integer) get(field);
}
@@ -387,6 +399,14 @@ public class Struct {
return set(def.name, value);
}
+ public Struct set(Field.Uint16 def, int value) {
+ if (value < 0 || value > 65535) {
+ throw new RuntimeException("Invalid value for unsigned short for " +
+ def.name + ": " + value);
+ }
+ return set(def.name, value);
+ }
+
public Struct set(Field.Float64 def, double 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 43a77fc..d801d8a 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
@@ -220,6 +220,44 @@ public abstract class Type {
}
};
+ public static final DocumentedType UINT16 = new DocumentedType() {
+ @Override
+ public void write(ByteBuffer buffer, Object o) {
+ Integer value = (Integer) o;
+ buffer.putShort((short) value.intValue());
+ }
+
+ @Override
+ public Object read(ByteBuffer buffer) {
+ short value = buffer.getShort();
+ return Integer.valueOf(Short.toUnsignedInt(value));
+ }
+
+ @Override
+ public int sizeOf(Object o) {
+ return 2;
+ }
+
+ @Override
+ public String typeName() {
+ return "UINT16";
+ }
+
+ @Override
+ public Integer validate(Object item) {
+ if (item instanceof Integer)
+ return (Integer) item;
+ else
+ throw new SchemaException(item + " is not an unsigned short.");
+ }
+
+ @Override
+ public String documentation() {
+ return "Represents an integer between 0 and 65535 inclusive. " +
+ "The values are encoded using two bytes in network byte order (big-endian).";
+ }
+ };
+
public static final DocumentedType INT32 = new DocumentedType() {
@Override
public void write(ByteBuffer buffer, Object o) {
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 5688d64..ada16d2 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
@@ -231,6 +231,18 @@ public class SimpleExampleMessageTest {
}
@Test
+ public void testMyUint16() {
+ // Verify that the uint16 field reads as 33000 when not set.
+ testRoundTrip(new SimpleExampleMessageData(),
+ message -> assertEquals(33000, message.myUint16()));
+
+ testRoundTrip(new SimpleExampleMessageData().setMyUint16(123),
+ message -> assertEquals(123, message.myUint16()));
+ testRoundTrip(new SimpleExampleMessageData().setMyUint16(60000),
+ message -> assertEquals(60000, message.myUint16()));
+ }
+
+ @Test
public void testMyString() {
// Verify that the tagged field reads as empty when not set.
testRoundTrip(new SimpleExampleMessageData(),
@@ -242,6 +254,11 @@ public class SimpleExampleMessageTest {
@Test
public void testMyBytes() {
+ assertThrows(RuntimeException.class,
+ () -> new SimpleExampleMessageData().setMyUint16(-1));
+ assertThrows(RuntimeException.class,
+ () -> new SimpleExampleMessageData().setMyUint16(65536));
+
// Verify that the tagged field reads as empty when not set.
testRoundTrip(new SimpleExampleMessageData(),
message -> assertArrayEquals(new byte[0], message.myBytes()));
diff --git a/clients/src/test/resources/common/message/SimpleExampleMessage.json b/clients/src/test/resources/common/message/SimpleExampleMessage.json
index 0a0fb1f..1fd6bdc 100644
--- a/clients/src/test/resources/common/message/SimpleExampleMessage.json
+++ b/clients/src/test/resources/common/message/SimpleExampleMessage.json
@@ -51,7 +51,8 @@
{ "name": "structId", "type": "string", "versions": "2+", "about": "String field in struct"}
]},
{ "name": "myCommonStruct", "type": "TestCommonStruct", "versions": "0+"},
- { "name": "myOtherCommonStruct", "type": "TestCommonStruct", "versions": "0+"}
+ { "name": "myOtherCommonStruct", "type": "TestCommonStruct", "versions": "0+"},
+ { "name": "myUint16", "type": "uint16", "versions": "1+", "default": "33000" }
],
"commonStructs": [
{ "name": "TestCommonStruct", "versions": "0+", "fields": [
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 3d1c677..0f31ba0 100644
--- a/generator/src/main/java/org/apache/kafka/message/FieldSpec.java
+++ b/generator/src/main/java/org/apache/kafka/message/FieldSpec.java
@@ -299,6 +299,7 @@ public final class FieldSpec {
}
} else if ((type instanceof FieldType.Int8FieldType) ||
(type instanceof FieldType.Int16FieldType) ||
+ (type instanceof FieldType.Uint16FieldType) ||
(type instanceof FieldType.Int32FieldType) ||
(type instanceof FieldType.Int64FieldType)) {
int base = 10;
@@ -331,6 +332,22 @@ public final class FieldSpec {
}
return "(short) " + fieldDefault;
}
+ } else if (type instanceof FieldType.Uint16FieldType) {
+ if (defaultString.isEmpty()) {
+ return "0";
+ } else {
+ try {
+ int value = Integer.valueOf(defaultString, base);
+ if (value < 0 || value > 65535) {
+ throw new RuntimeException("Invalid default for uint16 field " +
+ name + ": out of range.");
+ }
+ } catch (NumberFormatException e) {
+ throw new RuntimeException("Invalid default for uint16 field " +
+ name + ": " + defaultString, e);
+ }
+ return fieldDefault;
+ }
} else if (type instanceof FieldType.Int32FieldType) {
if (defaultString.isEmpty()) {
return "0";
@@ -457,6 +474,8 @@ public final class FieldSpec {
return "byte";
} else if (type instanceof FieldType.Int16FieldType) {
return "short";
+ } else if (type instanceof FieldType.Uint16FieldType) {
+ return "int";
} else if (type instanceof FieldType.Int32FieldType) {
return "int";
} else if (type instanceof FieldType.Int64FieldType) {
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 bb872f8..e0009c2 100644
--- a/generator/src/main/java/org/apache/kafka/message/FieldType.java
+++ b/generator/src/main/java/org/apache/kafka/message/FieldType.java
@@ -82,6 +82,26 @@ public interface FieldType {
}
}
+ final class Uint16FieldType implements FieldType {
+ static final Uint16FieldType INSTANCE = new Uint16FieldType();
+ private static final String NAME = "uint16";
+
+ @Override
+ public String getBoxedJavaType(HeaderGenerator headerGenerator) {
+ return "Integer";
+ }
+
+ @Override
+ public Optional<Integer> fixedLength() {
+ return Optional.of(2);
+ }
+
+ @Override
+ public String toString() {
+ return NAME;
+ }
+ }
+
final class Int32FieldType implements FieldType {
static final Int32FieldType INSTANCE = new Int32FieldType();
private static final String NAME = "int32";
@@ -347,6 +367,8 @@ public interface FieldType {
return Int8FieldType.INSTANCE;
case Int16FieldType.NAME:
return Int16FieldType.INSTANCE;
+ case Uint16FieldType.NAME:
+ return Uint16FieldType.INSTANCE;
case Int32FieldType.NAME:
return Int32FieldType.INSTANCE;
case Int64FieldType.NAME:
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 56809f4..e318876 100644
--- a/generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
@@ -159,6 +159,11 @@ public final class JsonConverterGenerator implements MessageClassGenerator {
buffer.printf("%s;%n", target.assignmentStatement(
String.format("MessageUtil.jsonNodeToShort(%s, \"%s\")",
target.sourceVariable(), target.humanReadableName())));
+ } else if (target.field().type() instanceof FieldType.Uint16FieldType) {
+ headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
+ buffer.printf("%s;%n", target.assignmentStatement(
+ String.format("MessageUtil.jsonNodeToUnsignedShort(%s, \"%s\")",
+ target.sourceVariable(), target.humanReadableName())));
} else if (target.field().type() instanceof FieldType.Int32FieldType) {
headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
buffer.printf("%s;%n", target.assignmentStatement(
@@ -332,11 +337,12 @@ public final class JsonConverterGenerator implements MessageClassGenerator {
buffer.printf("%s;%n", target.assignmentStatement(
String.format("BooleanNode.valueOf(%s)", target.sourceVariable())));
} else if ((target.field().type() instanceof FieldType.Int8FieldType) ||
- (target.field().type() instanceof FieldType.Int16FieldType)) {
+ (target.field().type() instanceof FieldType.Int16FieldType)) {
headerGenerator.addImport(MessageGenerator.SHORT_NODE_CLASS);
buffer.printf("%s;%n", target.assignmentStatement(
String.format("new ShortNode(%s)", target.sourceVariable())));
- } else if (target.field().type() instanceof FieldType.Int32FieldType) {
+ } else if ((target.field().type() instanceof FieldType.Int32FieldType) ||
+ (target.field().type() instanceof FieldType.Uint16FieldType)) {
headerGenerator.addImport(MessageGenerator.INT_NODE_CLASS);
buffer.printf("%s;%n", target.assignmentStatement(
String.format("new IntNode(%s)", target.sourceVariable())));
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 1d51585..e34c784 100644
--- a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
@@ -554,6 +554,8 @@ public final class MessageDataGenerator implements MessageClassGenerator {
return "_readable.readByte()";
} else if (type instanceof FieldType.Int16FieldType) {
return "_readable.readShort()";
+ } else if (type instanceof FieldType.Uint16FieldType) {
+ return "_readable.readUnsignedShort()";
} else if (type instanceof FieldType.Int32FieldType) {
return "_readable.readInt()";
} else if (type instanceof FieldType.Int64FieldType) {
@@ -826,6 +828,8 @@ public final class MessageDataGenerator implements MessageClassGenerator {
return String.format("struct.getByte(\"%s\")", name);
} else if (type instanceof FieldType.Int16FieldType) {
return String.format("struct.getShort(\"%s\")", name);
+ } else if (type instanceof FieldType.Uint16FieldType) {
+ return String.format("struct.getUnsignedShort(\"%s\")", name);
} else if (type instanceof FieldType.Int32FieldType) {
return String.format("struct.getInt(\"%s\")", name);
} else if (type instanceof FieldType.Int64FieldType) {
@@ -1039,6 +1043,8 @@ public final class MessageDataGenerator implements MessageClassGenerator {
return String.format("_writable.writeByte(%s)", name);
} else if (type instanceof FieldType.Int16FieldType) {
return String.format("_writable.writeShort(%s)", name);
+ } else if (type instanceof FieldType.Uint16FieldType) {
+ return String.format("_writable.writeUnsignedShort(%s)", name);
} else if (type instanceof FieldType.Int32FieldType) {
return String.format("_writable.writeInt(%s)", name);
} else if (type instanceof FieldType.Int64FieldType) {
@@ -1230,6 +1236,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
if ((field.type() instanceof FieldType.BoolFieldType) ||
(field.type() instanceof FieldType.Int8FieldType) ||
(field.type() instanceof FieldType.Int16FieldType) ||
+ (field.type() instanceof FieldType.Uint16FieldType) ||
(field.type() instanceof FieldType.Int32FieldType) ||
(field.type() instanceof FieldType.Int64FieldType) ||
(field.type() instanceof FieldType.UUIDFieldType) ||
@@ -1276,6 +1283,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
if ((field.type() instanceof FieldType.BoolFieldType) ||
(field.type() instanceof FieldType.Int8FieldType) ||
(field.type() instanceof FieldType.Int16FieldType) ||
+ (field.type() instanceof FieldType.Uint16FieldType) ||
(field.type() instanceof FieldType.Int32FieldType) ||
(field.type() instanceof FieldType.Int64FieldType) ||
(field.type() instanceof FieldType.UUIDFieldType) ||
@@ -1726,6 +1734,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
field.camelCaseName());
} else if ((field.type() instanceof FieldType.Int8FieldType) ||
(field.type() instanceof FieldType.Int16FieldType) ||
+ (field.type() instanceof FieldType.Uint16FieldType) ||
(field.type() instanceof FieldType.Int32FieldType)) {
buffer.printf("hashCode = 31 * hashCode + %s;%n",
field.camelCaseName());
@@ -1783,6 +1792,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
if ((field.type() instanceof FieldType.BoolFieldType) ||
(field.type() instanceof FieldType.Int8FieldType) ||
(field.type() instanceof FieldType.Int16FieldType) ||
+ (field.type() instanceof FieldType.Uint16FieldType) ||
(field.type() instanceof FieldType.Int32FieldType) ||
(field.type() instanceof FieldType.Int64FieldType) ||
(field.type() instanceof FieldType.Float64FieldType) ||
@@ -1869,6 +1879,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
buffer.printf("+ \"%s%s=\" + (%s ? \"true\" : \"false\")%n", prefix, field.camelCaseName(), field.camelCaseName());
} else if ((field.type() instanceof FieldType.Int8FieldType) ||
(field.type() instanceof FieldType.Int16FieldType) ||
+ (field.type() instanceof FieldType.Uint16FieldType) ||
(field.type() instanceof FieldType.Int32FieldType) ||
(field.type() instanceof FieldType.Int64FieldType) ||
(field.type() instanceof FieldType.Float64FieldType)) {
@@ -1891,6 +1902,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
prefix, field.camelCaseName(), field.camelCaseName());
} else if (field.type().isStruct() ||
field.type() instanceof FieldType.UUIDFieldType) {
+ } else if (field.type().isStruct()) {
buffer.printf("+ \"%s%s=\" + %s.toString()%n",
prefix, field.camelCaseName(), field.camelCaseName());
} else if (field.type().isArray()) {
@@ -1930,6 +1942,14 @@ public final class MessageDataGenerator implements MessageClassGenerator {
field.capitalizedCamelCaseName(),
field.fieldAbstractJavaType(headerGenerator, structRegistry));
buffer.incrementIndent();
+ if (field.type() instanceof FieldType.Uint16FieldType) {
+ buffer.printf("if (v < 0 || v > 65535) {%n");
+ buffer.incrementIndent();
+ buffer.printf("throw new RuntimeException(\"Invalid value \" + v + " +
+ "\"for unsigned short field.\");%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
buffer.printf("this.%s = v;%n", field.camelCaseName());
buffer.printf("return this;%n");
buffer.decrementIndent();
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 3b357a5..b6fb0aa 100644
--- a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
@@ -51,6 +51,10 @@ public final class MessageGenerator {
static final String API_MESSAGE_TYPE_JAVA = "ApiMessageType.java";
+ static final String METADATA_RECORD_TYPE_JAVA = "MetadataRecordType.java";
+
+ static final String METADATA_JSON_CONVERTERS_JAVA = "MetadataJsonConverters.java";
+
static final String API_MESSAGE_CLASS = "org.apache.kafka.common.protocol.ApiMessage";
static final String MESSAGE_CLASS = "org.apache.kafka.common.protocol.Message";
@@ -163,15 +167,26 @@ public final class MessageGenerator {
JSON_SERDE.setSerializationInclusion(JsonInclude.Include.NON_EMPTY);
}
- private static TypeClassGenerator createTypeClassGenerator(String packageName,
- String type) {
- if (type == null) return null;
- switch (type) {
- case "ApiMessageTypeGenerator":
- return new ApiMessageTypeGenerator(packageName);
- default:
- throw new RuntimeException("Unknown type class generator type '" + type + "'");
+ private static List<TypeClassGenerator> createTypeClassGenerators(String packageName,
+ List<String> types) {
+ if (types == null) return Collections.emptyList();
+ List<TypeClassGenerator> generators = new ArrayList<>();
+ for (String type : types) {
+ switch (type) {
+ case "ApiMessageTypeGenerator":
+ generators.add(new ApiMessageTypeGenerator(packageName));
+ break;
+ case "MetadataRecordTypeGenerator":
+ generators.add(new MetadataRecordTypeGenerator(packageName));
+ break;
+ case "MetadataJsonConvertersGenerator":
+ generators.add(new MetadataJsonConvertersGenerator(packageName));
+ break;
+ default:
+ throw new RuntimeException("Unknown type class generator type '" + type + "'");
+ }
}
+ return generators;
}
private static List<MessageClassGenerator> createMessageClassGenerators(String packageName,
@@ -196,12 +211,13 @@ public final class MessageGenerator {
public static void processDirectories(String packageName,
String outputDir,
String inputDir,
- String typeClassGeneratorType,
+ List<String> typeClassGeneratorTypes,
List<String> messageClassGeneratorTypes) throws Exception {
Files.createDirectories(Paths.get(outputDir));
int numProcessed = 0;
- TypeClassGenerator typeClassGenerator =
- createTypeClassGenerator(packageName, typeClassGeneratorType);
+
+ List<TypeClassGenerator> typeClassGenerators =
+ createTypeClassGenerators(packageName, typeClassGeneratorTypes);
HashSet<String> outputFileNames = new HashSet<>();
try (DirectoryStream<Path> directoryStream = Files
.newDirectoryStream(Paths.get(inputDir), JSON_GLOB)) {
@@ -220,15 +236,13 @@ public final class MessageGenerator {
}
}
numProcessed++;
- if (typeClassGenerator != null) {
- typeClassGenerator.registerMessageType(spec);
- }
+ typeClassGenerators.forEach(generator -> generator.registerMessageType(spec));
} catch (Exception e) {
throw new RuntimeException("Exception while processing " + inputPath.toString(), e);
}
}
}
- if (typeClassGenerator != null) {
+ for (TypeClassGenerator typeClassGenerator : typeClassGenerators) {
outputFileNames.add(typeClassGenerator.outputName());
Path factoryOutputPath = Paths.get(outputDir, typeClassGenerator.outputName());
try (BufferedWriter writer = Files.newBufferedWriter(factoryOutputPath)) {
@@ -332,10 +346,11 @@ public final class MessageGenerator {
.required(true)
.metavar("INPUT")
.help("The input directory to use.");
- parser.addArgument("--typeclass-generator", "-t")
+ parser.addArgument("--typeclass-generators", "-t")
+ .nargs("+")
.action(store())
- .metavar("TYPECLASS_GENERATOR")
- .help("The type class generator to use, if any.");
+ .metavar("TYPECLASS_GENERATORS")
+ .help("The type class generators to use, if any.");
parser.addArgument("--message-class-generators", "-m")
.nargs("+")
.action(store())
@@ -343,7 +358,7 @@ public final class MessageGenerator {
.help("The message class generators to use.");
Namespace res = parser.parseArgsOrFail(args);
processDirectories(res.getString("package"), res.getString("output"),
- res.getString("input"), res.getString("typeclass_generator"),
+ res.getString("input"), res.getList("typeclass_generators"),
res.getList("message_class_generators"));
}
}
diff --git a/generator/src/main/java/org/apache/kafka/message/MessageSpecType.java b/generator/src/main/java/org/apache/kafka/message/MessageSpecType.java
index d24108f..d1110d9 100644
--- a/generator/src/main/java/org/apache/kafka/message/MessageSpecType.java
+++ b/generator/src/main/java/org/apache/kafka/message/MessageSpecType.java
@@ -20,15 +20,33 @@ package org.apache.kafka.message;
import com.fasterxml.jackson.annotation.JsonProperty;
public enum MessageSpecType {
+ /**
+ * Kafka request RPCs.
+ */
@JsonProperty("request")
REQUEST,
+ /**
+ * Kafka response RPCs.
+ */
@JsonProperty("response")
RESPONSE,
+ /**
+ * Kafka RPC headers.
+ */
@JsonProperty("header")
HEADER,
+ /**
+ * KIP-631 controller records.
+ */
+ @JsonProperty("metadata")
+ METADATA,
+
+ /**
+ * Other message spec types.
+ */
@JsonProperty("data")
DATA;
}
diff --git a/generator/src/main/java/org/apache/kafka/message/MetadataJsonConvertersGenerator.java b/generator/src/main/java/org/apache/kafka/message/MetadataJsonConvertersGenerator.java
new file mode 100644
index 0000000..13321f0
--- /dev/null
+++ b/generator/src/main/java/org/apache/kafka/message/MetadataJsonConvertersGenerator.java
@@ -0,0 +1,123 @@
+/*
+ * 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.message;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.util.Map;
+import java.util.TreeMap;
+
+public class MetadataJsonConvertersGenerator implements TypeClassGenerator {
+ private final HeaderGenerator headerGenerator;
+ private final CodeBuffer buffer;
+ private final TreeMap<Short, MessageSpec> apis;
+
+ public MetadataJsonConvertersGenerator(String packageName) {
+ this.headerGenerator = new HeaderGenerator(packageName);
+ this.apis = new TreeMap<>();
+ this.buffer = new CodeBuffer();
+ }
+
+ @Override
+ public String outputName() {
+ return MessageGenerator.METADATA_JSON_CONVERTERS_JAVA;
+ }
+
+ @Override
+ public void registerMessageType(MessageSpec spec) {
+ if (spec.type() == MessageSpecType.METADATA) {
+ short id = spec.apiKey().get();
+ MessageSpec prevSpec = apis.put(id, spec);
+ if (prevSpec != null) {
+ throw new RuntimeException("Duplicate metadata record entry for type " +
+ id + ". Original claimant: " + prevSpec.name() + ". New " +
+ "claimant: " + spec.name());
+ }
+ }
+ }
+
+ @Override
+ public void generateAndWrite(BufferedWriter writer) throws IOException {
+ buffer.printf("public class MetadataJsonConverters {%n");
+ buffer.incrementIndent();
+ generateWriteJson();
+ buffer.printf("%n");
+ generateReadJson();
+ buffer.printf("%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ headerGenerator.generate();
+
+ headerGenerator.buffer().write(writer);
+ buffer.write(writer);
+ }
+
+ private void generateWriteJson() {
+ headerGenerator.addImport(MessageGenerator.JSON_NODE_CLASS);
+ headerGenerator.addImport(MessageGenerator.API_MESSAGE_CLASS);
+
+ buffer.printf("public static JsonNode writeJson(ApiMessage apiMessage, short apiVersion) {%n");
+ buffer.incrementIndent();
+ buffer.printf("switch (apiMessage.apiKey()) {%n");
+ buffer.incrementIndent();
+ for (Map.Entry<Short, MessageSpec> entry : apis.entrySet()) {
+ String apiMessageClassName = MessageGenerator.capitalizeFirst(entry.getValue().name());
+ buffer.printf("case %d:%n", entry.getKey());
+ buffer.incrementIndent();
+ buffer.printf("return %sJsonConverter.write((%s) apiMessage, apiVersion);%n", apiMessageClassName, apiMessageClassName);
+ buffer.decrementIndent();
+ }
+ buffer.printf("default:%n");
+ buffer.incrementIndent();
+ headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
+ buffer.printf("throw new UnsupportedVersionException(\"Unknown metadata id \"" +
+ " + apiMessage.apiKey());%n");
+ buffer.decrementIndent();
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+
+ private void generateReadJson() {
+ headerGenerator.addImport(MessageGenerator.JSON_NODE_CLASS);
+ headerGenerator.addImport(MessageGenerator.API_MESSAGE_CLASS);
+
+ buffer.printf("public static ApiMessage readJson(JsonNode json, short apiKey, short apiVersion) {%n");
+ buffer.incrementIndent();
+ buffer.printf("switch (apiKey) {%n");
+ buffer.incrementIndent();
+ for (Map.Entry<Short, MessageSpec> entry : apis.entrySet()) {
+ String apiMessageClassName = MessageGenerator.capitalizeFirst(entry.getValue().name());
+ buffer.printf("case %d:%n", entry.getKey());
+ buffer.incrementIndent();
+ buffer.printf("return %sJsonConverter.read(json, apiVersion);%n", apiMessageClassName);
+ buffer.decrementIndent();
+ }
+ buffer.printf("default:%n");
+ buffer.incrementIndent();
+ headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
+ buffer.printf("throw new UnsupportedVersionException(\"Unknown metadata id \"" +
+ " + apiKey);%n");
+ buffer.decrementIndent();
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+}
diff --git a/generator/src/main/java/org/apache/kafka/message/MetadataRecordTypeGenerator.java b/generator/src/main/java/org/apache/kafka/message/MetadataRecordTypeGenerator.java
new file mode 100644
index 0000000..6d70631
--- /dev/null
+++ b/generator/src/main/java/org/apache/kafka/message/MetadataRecordTypeGenerator.java
@@ -0,0 +1,180 @@
+/*
+ * 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.message;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+
+public final class MetadataRecordTypeGenerator implements TypeClassGenerator {
+ private final HeaderGenerator headerGenerator;
+ private final CodeBuffer buffer;
+ private final TreeMap<Short, MessageSpec> apis;
+
+ public MetadataRecordTypeGenerator(String packageName) {
+ this.headerGenerator = new HeaderGenerator(packageName);
+ this.apis = new TreeMap<>();
+ this.buffer = new CodeBuffer();
+ }
+
+ @Override
+ public String outputName() {
+ return MessageGenerator.METADATA_RECORD_TYPE_JAVA;
+ }
+
+ @Override
+ public void registerMessageType(MessageSpec spec) {
+ if (spec.type() == MessageSpecType.METADATA) {
+ short id = spec.apiKey().get();
+ MessageSpec prevSpec = apis.put(id, spec);
+ if (prevSpec != null) {
+ throw new RuntimeException("Duplicate metadata record entry for type " +
+ id + ". Original claimant: " + prevSpec.name() + ". New " +
+ "claimant: " + spec.name());
+ }
+ }
+ }
+
+ @Override
+ public void generateAndWrite(BufferedWriter writer) throws IOException {
+ generate();
+ write(writer);
+ }
+
+ private void generate() {
+ buffer.printf("public enum MetadataRecordType {%n");
+ buffer.incrementIndent();
+ generateEnumValues();
+ buffer.printf("%n");
+ generateInstanceVariables();
+ buffer.printf("%n");
+ generateEnumConstructor();
+ buffer.printf("%n");
+ generateFromApiKey();
+ buffer.printf("%n");
+ generateNewMetadataRecord();
+ buffer.printf("%n");
+ generateAccessor("id", "short");
+ buffer.printf("%n");
+ generateToString();
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ headerGenerator.generate();
+ }
+
+ private void generateEnumValues() {
+ int numProcessed = 0;
+ for (Map.Entry<Short, MessageSpec> entry : apis.entrySet()) {
+ MessageSpec spec = entry.getValue();
+ String name = spec.name();
+ numProcessed++;
+ buffer.printf("%s(\"%s\", (short) %d)%s%n",
+ MessageGenerator.toSnakeCase(name).toUpperCase(Locale.ROOT),
+ MessageGenerator.capitalizeFirst(name),
+ entry.getKey(),
+ (numProcessed == apis.size()) ? ";" : ",");
+ }
+ }
+
+ private void generateInstanceVariables() {
+ buffer.printf("private final String name;%n");
+ buffer.printf("private final short id;%n");
+ }
+
+ private void generateEnumConstructor() {
+ buffer.printf("MetadataRecordType(String name, short id) {%n");
+ buffer.incrementIndent();
+ buffer.printf("this.name = name;%n");
+ buffer.printf("this.id = id;%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+
+ private void generateFromApiKey() {
+ buffer.printf("public static MetadataRecordType fromId(short id) {%n");
+ buffer.incrementIndent();
+ buffer.printf("switch (id) {%n");
+ buffer.incrementIndent();
+ for (Map.Entry<Short, MessageSpec> entry : apis.entrySet()) {
+ buffer.printf("case %d:%n", entry.getKey());
+ buffer.incrementIndent();
+ buffer.printf("return %s;%n", MessageGenerator.
+ toSnakeCase(entry.getValue().name()).toUpperCase(Locale.ROOT));
+ buffer.decrementIndent();
+ }
+ buffer.printf("default:%n");
+ buffer.incrementIndent();
+ headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
+ buffer.printf("throw new UnsupportedVersionException(\"Unknown metadata id \"" +
+ " + id);%n");
+ buffer.decrementIndent();
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+
+ private void generateNewMetadataRecord() {
+ headerGenerator.addImport(MessageGenerator.API_MESSAGE_CLASS);
+ buffer.printf("public ApiMessage newMetadataRecord() {%n");
+ buffer.incrementIndent();
+ buffer.printf("switch (id) {%n");
+ buffer.incrementIndent();
+ for (Map.Entry<Short, MessageSpec> entry : apis.entrySet()) {
+ buffer.printf("case %d:%n", entry.getKey());
+ buffer.incrementIndent();
+ buffer.printf("return new %s();%n",
+ MessageGenerator.capitalizeFirst(entry.getValue().name()));
+ buffer.decrementIndent();
+ }
+ buffer.printf("default:%n");
+ buffer.incrementIndent();
+ headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
+ buffer.printf("throw new UnsupportedVersionException(\"Unknown metadata id \"" +
+ " + id);%n");
+ buffer.decrementIndent();
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+
+ private void generateAccessor(String name, String type) {
+ buffer.printf("public %s %s() {%n", type, name);
+ buffer.incrementIndent();
+ buffer.printf("return this.%s;%n", name);
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+
+ private void generateToString() {
+ buffer.printf("@Override%n");
+ buffer.printf("public String toString() {%n");
+ buffer.incrementIndent();
+ buffer.printf("return this.name();%n");
+ buffer.decrementIndent();
+ buffer.printf("}%n");
+ }
+
+ private void write(BufferedWriter writer) throws IOException {
+ headerGenerator.buffer().write(writer);
+ buffer.write(writer);
+ }
+}
diff --git a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java
index 9a6b775..5ebd158 100644
--- a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java
+++ b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java
@@ -244,6 +244,12 @@ final class SchemaGenerator {
throw new RuntimeException("Type " + type + " cannot be nullable.");
}
return "Type.INT16";
+ } else if (type instanceof FieldType.Uint16FieldType) {
+ headerGenerator.addImport(MessageGenerator.TYPE_CLASS);
+ if (nullable) {
+ throw new RuntimeException("Type " + type + " cannot be nullable.");
+ }
+ return "Type.UINT16";
} else if (type instanceof FieldType.Int32FieldType) {
headerGenerator.addImport(MessageGenerator.TYPE_CLASS);
if (nullable) {