KAFKA-7609: Add Protocol Generator for Kafka

.gitignore

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

build.gradle

@@ -839,6 +839,7 @@ project(':clients') {
     testRuntime libs.slf4jlog4j
     testRuntime libs.jacksonDatabind
     testRuntime libs.jacksonJDK8Datatypes
+    testCompile libs.jacksonJaxrsJsonProvider
   }
 
   task determineCommitId {
@@ -887,6 +888,27 @@ project(':clients') {
     delete "$buildDir/kafka/"
   }
 
+  task processMessages(type:org.apache.kafka.task.ProcessMessagesTask) {
+    inputDirectory = file("src/main/resources/common/message")
+    outputDirectory = file("src/generated/java/org/apache/kafka/common/message")
+  }
+
+  sourceSets {
+    main {
+      java {
+        srcDirs = ["src/generated/java", "src/main/java"]
+      }
+    }
+    test {
+      java {
+        srcDirs = ["src/generated/java", "src/test/java",
+            "$rootDir/buildSrc/src/main/java/org/apache/kafka/message/"]
+      }
+    }
+  }
+
+  compileJava.dependsOn 'processMessages'
+
   javadoc {
     include "**/org/apache/kafka/clients/admin/*"
     include "**/org/apache/kafka/clients/consumer/*"

buildSrc/build.gradle

@@ -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.
+
+repositories {
+  mavenCentral()
+}
+
+dependencies {
+  compile "com.fasterxml.jackson.core:jackson-databind:2.9.6"
+  compile "com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.9.6"
+}
+
+// Workaround for https://github.com/gradle/gradle/issues/1271
+test.enabled=false

buildSrc/src/main/java/org/apache/kafka/message/CodeBuffer.java

@@ -0,0 +1,81 @@
+/*
+ * 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.IOException;
+import java.io.Writer;
+import java.util.ArrayList;
+
+public class CodeBuffer {
+    private final ArrayList<String> lines;
+    private int indent;
+
+    public CodeBuffer() {
+        this.lines = new ArrayList<>();
+        this.indent = 0;
+    }
+
+    public void incrementIndent() {
+        indent++;
+    }
+
+    public void decrementIndent() {
+        indent--;
+        if (indent < 0) {
+            throw new RuntimeException("Indent < 0");
+        }
+    }
+
+    public void printf(String format, Object... args) {
+        lines.add(String.format(indentSpaces() + format, args));
+    }
+
+    public void write(Writer writer) throws IOException {
+        for (String line : lines) {
+            writer.write(line);
+        }
+    }
+
+    public void write(CodeBuffer other) {
+        for (String line : lines) {
+            other.lines.add(other.indentSpaces() + line);
+        }
+    }
+
+    private String indentSpaces() {
+        StringBuilder bld = new StringBuilder();
+        for (int i = 0; i < indent; i++) {
+            bld.append("    ");
+        }
+        return bld.toString();
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (!(other instanceof CodeBuffer)) {
+            return false;
+        }
+        CodeBuffer o = (CodeBuffer) other;
+        return lines.equals(o.lines);
+    }
+
+    @Override
+    public int hashCode() {
+        return lines.hashCode();
+    }
+}

buildSrc/src/main/java/org/apache/kafka/message/FieldSpec.java

@@ -0,0 +1,142 @@
+/*
+ * 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 com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+import java.util.Objects;
+
+public final class FieldSpec {
+    private final StructSpec struct;
+
+    private final FieldType type;
+
+    private final boolean mapKey;
+
+    private final Versions nullableVersions;
+
+    private final String fieldDefault;
+
+    private final boolean ignorable;
+
+    private final String about;
+
+    @JsonCreator
+    public FieldSpec(@JsonProperty("name") String name,
+                     @JsonProperty("versions") String versions,
+                     @JsonProperty("fields") List<FieldSpec> fields,
+                     @JsonProperty("type") String type,
+                     @JsonProperty("mapKey") boolean mapKey,
+                     @JsonProperty("nullableVersions") String nullableVersions,
+                     @JsonProperty("default") String fieldDefault,
+                     @JsonProperty("ignorable") boolean ignorable,
+                     @JsonProperty("about") String about) {
+        this.struct = new StructSpec(name, versions, fields);
+        this.type = FieldType.parse(Objects.requireNonNull(type));
+        this.mapKey = mapKey;
+        this.nullableVersions = Versions.parse(nullableVersions, Versions.NONE);
+        if (!this.nullableVersions.empty()) {
+            if (!this.type.canBeNullable()) {
+                throw new RuntimeException("Type " + this.type + " cannot be nullable.");
+            }
+        }
+        this.fieldDefault = fieldDefault == null ? "" : fieldDefault;
+        this.ignorable = ignorable;
+        this.about = about == null ? "" : about;
+        if (!this.struct.fields().isEmpty()) {
+            if (!this.type.isArray()) {
+                throw new RuntimeException("Non-array field " + name + " cannot have fields");
+            }
+        }
+    }
+
+    public StructSpec struct() {
+        return struct;
+    }
+
+    @JsonProperty("name")
+    public String name() {
+        return struct.name();
+    }
+
+    String capitalizedCamelCaseName() {
+        return MessageGenerator.capitalizeFirst(struct.name());
+    }
+
+    String camelCaseName() {
+        return MessageGenerator.lowerCaseFirst(struct.name());
+    }
+
+    String snakeCaseName() {
+        return MessageGenerator.toSnakeCase(struct.name());
+    }
+
+    @JsonProperty("versions")
+    public String versionsString() {
+        return struct.versionsString();
+    }
+
+    @JsonProperty("fields")
+    public List<FieldSpec> fields() {
+        return struct.fields();
+    }
+
+    @JsonProperty("type")
+    public String typeString() {
+        return type.toString();
+    }
+
+    public FieldType type() {
+        return type;
+    }
+
+    @JsonProperty("mapKey")
+    public boolean mapKey() {
+        return mapKey;
+    }
+
+    public Versions nullableVersions() {
+        return nullableVersions;
+    }
+
+    @JsonProperty("nullableVersions")
+    public String nullableVersionsString() {
+        return nullableVersions.toString();
+    }
+
+    @JsonProperty("default")
+    public String defaultString() {
+        return fieldDefault;
+    }
+
+    boolean hasKeys() {
+        return struct.hasKeys();
+    }
+
+    @JsonProperty("ignorable")
+    public boolean ignorable() {
+        return ignorable;
+    }
+
+    @JsonProperty("about")
+    public String about() {
+        return about;
+    }
+}