KAFKA-2945: CreateTopic - protocol and server side implementation

core/src/main/scala/kafka/common/TopicExistsException.scala → clients/src/main/java/org/apache/kafka/common/errors/InvalidConfigurationException.java

@@ -6,17 +6,26 @@
  * (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
+ *      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.errors;
 
-package kafka.common
+public class InvalidConfigurationException extends ApiException {
 
-class TopicExistsException(message: String) extends RuntimeException(message) {
-  def this() = this(null) 
-}
+    private static final long serialVersionUID = 1L;
+
+    public InvalidConfigurationException(String message) {
+        super(message);
+    }
+
+    public InvalidConfigurationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

core/src/main/scala/kafka/common/InvalidTopicException.scala → clients/src/main/java/org/apache/kafka/common/errors/InvalidPartitionsException.java

@@ -6,17 +6,26 @@
  * (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
+ *      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.errors;
 
-package kafka.common
+public class InvalidPartitionsException extends ApiException {
 
-class InvalidTopicException(message: String) extends RuntimeException(message) {
-  def this() = this(null) 
-}
+    private static final long serialVersionUID = 1L;
+
+    public InvalidPartitionsException(String message) {
+        super(message);
+    }
+
+    public InvalidPartitionsException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicaAssignmentException.java

@@ -0,0 +1,31 @@
+/**
+ * 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.errors;
+
+public class InvalidReplicaAssignmentException extends ApiException {
+
+    private static final long serialVersionUID = 1L;
+
+    public InvalidReplicaAssignmentException(String message) {
+        super(message);
+    }
+
+    public InvalidReplicaAssignmentException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicationFactorException.java

@@ -0,0 +1,31 @@
+/**
+ * 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.errors;
+
+public class InvalidReplicationFactorException extends ApiException {
+
+    private static final long serialVersionUID = 1L;
+
+    public InvalidReplicationFactorException(String message) {
+        super(message);
+    }
+
+    public InvalidReplicationFactorException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/errors/InvalidRequestException.java

@@ -0,0 +1,36 @@
+/**
+ * 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.errors;
+
+/**
+ * Thrown when a request breaks basic wire protocol rules.
+ * This most likely occurs because of a request being malformed by the client library or
+ * the message was sent to an incompatible broker.
+ */
+public class InvalidRequestException extends ApiException {
+
+    private static final long serialVersionUID = 1L;
+
+    public InvalidRequestException(String message) {
+        super(message);
+    }
+
+    public InvalidRequestException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

clients/src/main/java/org/apache/kafka/common/errors/NotControllerException.java

@@ -0,0 +1,28 @@
+/**
+ * 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.errors;
+
+public class NotControllerException extends RetriableException {
+
+    private static final long serialVersionUID = 1L;
+
+    public NotControllerException(String message) {
+        super(message);
+    }
+
+    public NotControllerException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

core/src/main/scala/kafka/network/InvalidRequestException.scala → clients/src/main/java/org/apache/kafka/common/errors/TopicExistsException.java

@@ -6,19 +6,26 @@
  * (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
+ *      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.errors;
 
-package kafka.network
+public class TopicExistsException extends ApiException {
 
-class InvalidRequestException(val message: String, cause: Throwable) extends RuntimeException(message, cause) {
+    private static final long serialVersionUID = 1L;
+
+    public TopicExistsException(String message) {
+        super(message);
+    }
+
+    public TopicExistsException(String message, Throwable cause) {
+        super(message, cause);
+    }
 
-  def this() = this("", null)
-  def this(message: String) = this(message, null)
 }

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

@@ -38,7 +38,8 @@ public enum ApiKeys {
     DESCRIBE_GROUPS(15, "DescribeGroups"),
     LIST_GROUPS(16, "ListGroups"),
     SASL_HANDSHAKE(17, "SaslHandshake"),
-    API_VERSIONS(18, "ApiVersions");
+    API_VERSIONS(18, "ApiVersions"),
+    CREATE_TOPICS(19, "CreateTopics");
 
     private static final ApiKeys[] ID_TO_TYPE;
     private static final int MIN_API_KEY = 0;
@@ -98,4 +99,4 @@ public static void main(String[] args) {
         System.out.println(toHtml());
     }
 
-}
+}

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

@@ -31,14 +31,20 @@
 import org.apache.kafka.common.errors.IllegalSaslStateException;
 import org.apache.kafka.common.errors.InconsistentGroupProtocolException;
 import org.apache.kafka.common.errors.InvalidCommitOffsetSizeException;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.apache.kafka.common.errors.InvalidFetchSizeException;
 import org.apache.kafka.common.errors.InvalidGroupIdException;
+import org.apache.kafka.common.errors.InvalidPartitionsException;
+import org.apache.kafka.common.errors.InvalidReplicaAssignmentException;
+import org.apache.kafka.common.errors.InvalidReplicationFactorException;
+import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.errors.InvalidRequiredAcksException;
 import org.apache.kafka.common.errors.InvalidSessionTimeoutException;
 import org.apache.kafka.common.errors.InvalidTimestampException;
 import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.LeaderNotAvailableException;
 import org.apache.kafka.common.errors.NetworkException;
+import org.apache.kafka.common.errors.NotControllerException;
 import org.apache.kafka.common.errors.NotCoordinatorForGroupException;
 import org.apache.kafka.common.errors.NotEnoughReplicasAfterAppendException;
 import org.apache.kafka.common.errors.NotEnoughReplicasException;
@@ -50,6 +56,7 @@
 import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.ReplicaNotAvailableException;
 import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.errors.UnsupportedSaslMechanismException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
@@ -139,7 +146,22 @@ public enum Errors {
     ILLEGAL_SASL_STATE(34,
             new IllegalSaslStateException("Request is not valid given the current SASL state.")),
     UNSUPPORTED_VERSION(35,
-            new UnsupportedVersionException("The version of API is not supported."));
+            new UnsupportedVersionException("The version of API is not supported.")),
+    TOPIC_ALREADY_EXISTS(36,
+            new TopicExistsException("Topic with this name already exists.")),
+    INVALID_PARTITIONS(37,
+            new InvalidPartitionsException("Number of partitions is invalid.")),
+    INVALID_REPLICATION_FACTOR(38,
+            new InvalidReplicationFactorException("Replication-factor is invalid.")),
+    INVALID_REPLICA_ASSIGNMENT(39,
+            new InvalidReplicaAssignmentException("Replica assignment is invalid.")),
+    INVALID_CONFIG(40,
+            new InvalidConfigurationException("Configuration is invalid.")),
+    NOT_CONTROLLER(41,
+        new NotControllerException("This is not the correct controller for this cluster.")),
+    INVALID_REQUEST(42,
+        new InvalidRequestException("This most likely occurs because of a request being malformed by the client library or" +
+            " the message was sent to an incompatible broker. See the broker logs for more details."));
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

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

@@ -770,6 +770,50 @@ public class Protocol {
     public static final Schema[] API_VERSIONS_REQUEST = new Schema[]{API_VERSIONS_REQUEST_V0};
     public static final Schema[] API_VERSIONS_RESPONSE = new Schema[]{API_VERSIONS_RESPONSE_V0};
 
+    /* Admin requests common */
+    public static final Schema CONFIG_ENTRY = new Schema(new Field("config_key", STRING, "Configuration key name"),
+        new Field("config_value", STRING, "Configuration value"));
+
+    public static final Schema PARTITION_REPLICA_ASSIGNMENT_ENTRY = new Schema(
+        new Field("partition_id", INT32),
+        new Field("replicas", new ArrayOf(INT32), "The set of all nodes that should host this partition. The first replica in the list is the preferred leader."));
+
+    public static final Schema TOPIC_ERROR_CODE = new Schema(new Field("topic", STRING), new Field("error_code", INT16));
+
+    /* CreateTopic api */
+    public static final Schema SINGLE_CREATE_TOPIC_REQUEST_V0 = new Schema(
+        new Field("topic",
+            STRING,
+            "Name for newly created topic."),
+        new Field("num_partitions",
+            INT32,
+            "Number of partitions to be created. -1 indicates unset."),
+        new Field("replication_factor",
+            INT16,
+            "Replication factor for the topic. -1 indicates unset."),
+        new Field("replica_assignment",
+            new ArrayOf(PARTITION_REPLICA_ASSIGNMENT_ENTRY),
+            "Replica assignment among kafka brokers for this topic partitions. If this is set num_partitions and replication_factor must be unset."),
+        new Field("configs",
+            new ArrayOf(CONFIG_ENTRY),
+            "Topic level configuration for topic to be set."));
+
+    public static final Schema CREATE_TOPICS_REQUEST_V0 = new Schema(
+        new Field("create_topic_requests",
+            new ArrayOf(SINGLE_CREATE_TOPIC_REQUEST_V0),
+            "An array of single topic creation requests. Can not have multiple entries for the same topic."),
+        new Field("timeout",
+            INT32,
+            "The time in ms to wait for a topic to be completely created on the controller node. Values <= 0 will trigger topic creation and return immediately"));
+
+    public static final Schema CREATE_TOPICS_RESPONSE_V0 = new Schema(
+        new Field("topic_error_codes",
+            new ArrayOf(TOPIC_ERROR_CODE),
+            "An array of per topic error codes."));
+
+    public static final Schema[] CREATE_TOPICS_REQUEST = new Schema[] {CREATE_TOPICS_REQUEST_V0};
+    public static final Schema[] CREATE_TOPICS_RESPONSE = new Schema[] {CREATE_TOPICS_RESPONSE_V0};
+
     /* an array of all requests and responses with all schema versions; a null value in the inner array means that the
      * particular version is not supported */
     public static final Schema[][] REQUESTS = new Schema[ApiKeys.MAX_API_KEY + 1][];
@@ -799,6 +843,7 @@ public class Protocol {
         REQUESTS[ApiKeys.LIST_GROUPS.id] = LIST_GROUPS_REQUEST;
         REQUESTS[ApiKeys.SASL_HANDSHAKE.id] = SASL_HANDSHAKE_REQUEST;
         REQUESTS[ApiKeys.API_VERSIONS.id] = API_VERSIONS_REQUEST;
+        REQUESTS[ApiKeys.CREATE_TOPICS.id] = CREATE_TOPICS_REQUEST;
 
         RESPONSES[ApiKeys.PRODUCE.id] = PRODUCE_RESPONSE;
         RESPONSES[ApiKeys.FETCH.id] = FETCH_RESPONSE;
@@ -819,6 +864,7 @@ public class Protocol {
         RESPONSES[ApiKeys.LIST_GROUPS.id] = LIST_GROUPS_RESPONSE;
         RESPONSES[ApiKeys.SASL_HANDSHAKE.id] = SASL_HANDSHAKE_RESPONSE;
         RESPONSES[ApiKeys.API_VERSIONS.id] = API_VERSIONS_RESPONSE;
+        RESPONSES[ApiKeys.CREATE_TOPICS.id] = CREATE_TOPICS_RESPONSE;
 
         /* set the minimum and maximum version of each api */
         for (ApiKeys api : ApiKeys.values()) {

clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java

@@ -76,9 +76,11 @@ public static AbstractRequest getRequest(int requestId, int versionId, ByteBuffe
                 return SaslHandshakeRequest.parse(buffer, versionId);
             case API_VERSIONS:
                 return ApiVersionsRequest.parse(buffer, versionId);
+            case CREATE_TOPICS:
+                return CreateTopicsRequest.parse(buffer, versionId);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `getRequest`, the " +
                         "code should be updated to do so.", apiKey));
         }
     }
-}
+}