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"
+  compile "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.9.6"
+}
+
+test.enabled=false

buildSrc/src/main/java/org/apache/kafka/message/ApiMessageFactoryGenerator.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 final class ApiMessageFactoryGenerator {
+    private final TreeMap<Short, String> requestApis;
+    private final TreeMap<Short, String> responseApis;
+    private final HeaderGenerator headerGenerator;
+    private final CodeBuffer buffer;
+
+    public void registerMessageType(MessageSpec spec) {
+        if (spec.type() == MessageSpecType.REQUEST) {
+            if (requestApis.containsKey(spec.apiKey().get())) {
+                throw new RuntimeException("Found more than one request with " +
+                    "API key " + spec.apiKey().get());
+            }
+            requestApis.put(spec.apiKey().get(), spec.generatedClassName());
+        } else if (spec.type() == MessageSpecType.RESPONSE) {
+            if (responseApis.containsKey(spec.apiKey().get())) {
+                throw new RuntimeException("Found more than one response with " +
+                    "API key " + spec.apiKey().get());
+            }
+            responseApis.put(spec.apiKey().get(), spec.generatedClassName());
+        }
+    }
+
+    public ApiMessageFactoryGenerator() {
+        this.requestApis = new TreeMap<>();
+        this.responseApis = new TreeMap<>();
+        this.headerGenerator = new HeaderGenerator();
+        this.buffer = new CodeBuffer();
+    }
+
+    public void generate() {
+        buffer.printf("public final class ApiMessageFactory {%n");
+        buffer.incrementIndent();
+        generateFactoryMethod("request", requestApis);
+        buffer.printf("%n");
+        generateFactoryMethod("response", responseApis);
+        buffer.printf("%n");
+        generateSchemasAccessor("request", requestApis);
+        buffer.printf("%n");
+        generateSchemasAccessor("response", responseApis);
+        buffer.decrementIndent();
+        buffer.printf("}%n");
+        headerGenerator.generate();
+    }
+
+    public void generateFactoryMethod(String type, TreeMap<Short, String> apis) {
+        headerGenerator.addImport(MessageGenerator.MESSAGE_CLASS);
+        buffer.printf("public static Message new%s(short apiKey) {%n",
+            MessageGenerator.capitalizeFirst(type));
+        buffer.incrementIndent();
+        buffer.printf("switch (apiKey) {%n");
+        buffer.incrementIndent();
+        for (Map.Entry<Short, String> entry : apis.entrySet()) {
+            buffer.printf("case %d:%n", entry.getKey());
+            buffer.incrementIndent();
+            buffer.printf("return new %s();%n", entry.getValue());
+            buffer.decrementIndent();
+        }
+        buffer.printf("default:%n");
+        buffer.incrementIndent();
+        headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
+        buffer.printf("throw new UnsupportedVersionException(\"Unsupported %s API key \"" +
+            " + apiKey);%n", type);
+        buffer.decrementIndent();
+        buffer.decrementIndent();
+        buffer.printf("}%n");
+        buffer.decrementIndent();
+        buffer.printf("}%n");
+    }
+
+    public void generateSchemasAccessor(String type, TreeMap<Short, String> apis) {
+        headerGenerator.addImport(MessageGenerator.SCHEMA_CLASS);
+        buffer.printf("public static Schema[] %sSchemas(short apiKey) {%n",
+            MessageGenerator.lowerCaseFirst(type));
+        buffer.incrementIndent();
+        buffer.printf("switch (apiKey) {%n");
+        buffer.incrementIndent();
+        for (Map.Entry<Short, String> entry : apis.entrySet()) {
+            buffer.printf("case %d:%n", entry.getKey());
+            buffer.incrementIndent();
+            buffer.printf("return %s.SCHEMAS;%n", entry.getValue());
+            buffer.decrementIndent();
+        }
+        buffer.printf("default:%n");
+        buffer.incrementIndent();
+        headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
+        buffer.printf("throw new UnsupportedVersionException(\"Unsupported %s API key \"" +
+            " + apiKey);%n", type);
+        buffer.decrementIndent();
+        buffer.decrementIndent();
+        buffer.printf("}%n");
+        buffer.decrementIndent();
+        buffer.printf("}%n");
+    }
+
+    public void write(BufferedWriter writer) throws IOException {
+        headerGenerator.buffer().write(writer);
+        buffer.write(writer);
+    }
+}

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();
+    }
+}