From 4e1f59ae8807442c89f8594ac46dfc59ceb93866 Mon Sep 17 00:00:00 2001 From: Grant Henke Date: Thu, 9 Jun 2016 16:34:14 -0500 Subject: [PATCH 1/5] KAFKA-2945: CreateTopic - protocol and server side implementation --- .../errors/InvalidConfigurationException.java | 19 +- .../errors/InvalidPartitionsException.java | 19 +- .../InvalidReplicaAssignmentException.java | 31 +++ .../InvalidReplicationFactorException.java | 31 +++ .../errors/InvalidRequestException.java | 36 +++ .../common/errors/NotControllerException.java | 28 +++ .../common/errors/TopicExistsException.java | 17 +- .../apache/kafka/common/protocol/ApiKeys.java | 5 +- .../apache/kafka/common/protocol/Errors.java | 24 +- .../kafka/common/protocol/Protocol.java | 46 ++++ .../common/requests/AbstractRequest.java | 4 +- .../common/requests/CreateTopicsRequest.java | 230 ++++++++++++++++++ .../common/requests/CreateTopicsResponse.java | 95 ++++++++ .../common/requests/RequestResponseTest.java | 30 ++- .../main/scala/kafka/admin/AdminUtils.scala | 19 +- .../main/scala/kafka/admin/TopicCommand.scala | 6 +- .../src/main/scala/kafka/cluster/Broker.scala | 3 - .../scala/kafka/common/ErrorMapping.scala | 1 + .../kafka/controller/KafkaController.scala | 12 + core/src/main/scala/kafka/log/LogConfig.scala | 7 +- .../scala/kafka/network/RequestChannel.scala | 5 +- .../scala/kafka/network/SocketServer.scala | 1 + .../scala/kafka/server/AdminManager.scala | 103 ++++++++ .../kafka/server/DelayedCreateTopics.scala | 94 +++++++ .../kafka/server/DelayedOperationKey.scala | 6 + .../main/scala/kafka/server/KafkaApis.scala | 43 +++- .../main/scala/kafka/server/KafkaServer.scala | 6 +- .../kafka/api/AuthorizerIntegrationTest.scala | 22 +- .../scala/unit/kafka/admin/AdminTest.scala | 12 +- .../unit/kafka/admin/TopicCommandTest.scala | 4 +- .../unit/kafka/server/BaseRequestTest.scala | 36 ++- .../server/CreateTopicsRequestTest.scala | 212 ++++++++++++++++ .../kafka/server/ProduceRequestTest.scala | 5 +- .../server/SaslApiVersionsRequestTest.scala | 4 +- 34 files changed, 1149 insertions(+), 67 deletions(-) rename core/src/main/scala/kafka/common/TopicExistsException.scala => clients/src/main/java/org/apache/kafka/common/errors/InvalidConfigurationException.java (64%) rename core/src/main/scala/kafka/common/InvalidTopicException.scala => clients/src/main/java/org/apache/kafka/common/errors/InvalidPartitionsException.java (64%) create mode 100644 clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicaAssignmentException.java create mode 100644 clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicationFactorException.java create mode 100644 clients/src/main/java/org/apache/kafka/common/errors/InvalidRequestException.java create mode 100644 clients/src/main/java/org/apache/kafka/common/errors/NotControllerException.java rename core/src/main/scala/kafka/network/InvalidRequestException.scala => clients/src/main/java/org/apache/kafka/common/errors/TopicExistsException.java (66%) create mode 100644 clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java create mode 100644 clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java create mode 100644 core/src/main/scala/kafka/server/AdminManager.scala create mode 100644 core/src/main/scala/kafka/server/DelayedCreateTopics.scala create mode 100644 core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala diff --git a/core/src/main/scala/kafka/common/TopicExistsException.scala b/clients/src/main/java/org/apache/kafka/common/errors/InvalidConfigurationException.java similarity index 64% rename from core/src/main/scala/kafka/common/TopicExistsException.scala rename to clients/src/main/java/org/apache/kafka/common/errors/InvalidConfigurationException.java index 88a084e810137..25cbc7ae6750e 100644 --- a/core/src/main/scala/kafka/common/TopicExistsException.scala +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidConfigurationException.java @@ -6,7 +6,7 @@ * (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, @@ -14,9 +14,18 @@ * 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) -} \ No newline at end of file + private static final long serialVersionUID = 1L; + + public InvalidConfigurationException(String message) { + super(message); + } + + public InvalidConfigurationException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/core/src/main/scala/kafka/common/InvalidTopicException.scala b/clients/src/main/java/org/apache/kafka/common/errors/InvalidPartitionsException.java similarity index 64% rename from core/src/main/scala/kafka/common/InvalidTopicException.scala rename to clients/src/main/java/org/apache/kafka/common/errors/InvalidPartitionsException.java index 59f887490d417..c7ea66801802d 100644 --- a/core/src/main/scala/kafka/common/InvalidTopicException.scala +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidPartitionsException.java @@ -6,7 +6,7 @@ * (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, @@ -14,9 +14,18 @@ * 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) -} \ No newline at end of file + private static final long serialVersionUID = 1L; + + public InvalidPartitionsException(String message) { + super(message); + } + + public InvalidPartitionsException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicaAssignmentException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicaAssignmentException.java new file mode 100644 index 0000000000000..765f0f6726ed9 --- /dev/null +++ b/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); + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicationFactorException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidReplicationFactorException.java new file mode 100644 index 0000000000000..33f048a532870 --- /dev/null +++ b/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); + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidRequestException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidRequestException.java new file mode 100644 index 0000000000000..8299da45e7b2a --- /dev/null +++ b/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); + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/errors/NotControllerException.java b/clients/src/main/java/org/apache/kafka/common/errors/NotControllerException.java new file mode 100644 index 0000000000000..c2784a880e8f5 --- /dev/null +++ b/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); + } + +} diff --git a/core/src/main/scala/kafka/network/InvalidRequestException.scala b/clients/src/main/java/org/apache/kafka/common/errors/TopicExistsException.java similarity index 66% rename from core/src/main/scala/kafka/network/InvalidRequestException.scala rename to clients/src/main/java/org/apache/kafka/common/errors/TopicExistsException.java index 47dba6cced4c8..0fc0683195229 100644 --- a/core/src/main/scala/kafka/network/InvalidRequestException.scala +++ b/clients/src/main/java/org/apache/kafka/common/errors/TopicExistsException.java @@ -6,7 +6,7 @@ * (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, @@ -14,11 +14,18 @@ * 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) } diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java index aeb0b453b2b06..bd00b971b088d 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java +++ b/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()); } -} \ No newline at end of file +} diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java index bd7310ba4551f..90aa8f88a229a 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java +++ b/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); diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java index ec744272aa03b..e2064e12e67aa 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java +++ b/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 immediatly")); + + 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()) { diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java index ab61c6634271e..6a91825a82d90 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java +++ b/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)); } } -} \ No newline at end of file +} diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java new file mode 100644 index 0000000000000..d5ce56e6988b7 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java @@ -0,0 +1,230 @@ +/** + * 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.requests; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.ProtoUtils; +import org.apache.kafka.common.protocol.types.Schema; +import org.apache.kafka.common.protocol.types.Struct; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +public class CreateTopicsRequest extends AbstractRequest { + private static final Schema CURRENT_SCHEMA = ProtoUtils.currentRequestSchema(ApiKeys.CREATE_TOPICS.id); + + private static final String REQUESTS_KEY_NAME = "create_topic_requests"; + + private static final String TIMEOUT_KEY_NAME = "timeout"; + private static final String TOPIC_KEY_NAME = "topic"; + private static final String NUM_PARTITIONS_KEY_NAME = "num_partitions"; + private static final String REPLICATION_FACTOR_KEY_NAME = "replication_factor"; + private static final String REPLICA_ASSIGNMENT_KEY_NAME = "replica_assignment"; + private static final String REPLICA_ASSIGNMENT_PARTITION_ID_KEY_NAME = "partition_id"; + private static final String REPLICA_ASSIGNMENT_REPLICAS_KEY_NAME = "replicas"; + + private static final String CONFIG_KEY_KEY_NAME = "config_key"; + private static final String CONFIG_VALUE_KEY_NAME = "config_value"; + private static final String CONFIGS_KEY_NAME = "configs"; + + public static final class TopicDetails { + public final int numPartitions; + public final short replicationFactor; + public final Map> replicasAssignments; + public final Map configs; + + private TopicDetails(int numPartitions, + short replicationFactor, + Map> replicasAssignments, + Map configs) { + this.numPartitions = numPartitions; + this.replicationFactor = replicationFactor; + this.replicasAssignments = replicasAssignments; + this.configs = configs; + } + + public TopicDetails(int partitions, + short replicationFactor, + Map configs) { + this(partitions, replicationFactor, new HashMap>(), configs); + } + + public TopicDetails(int partitions, + short replicationFactor) { + this(partitions, replicationFactor, new HashMap()); + } + + public TopicDetails(Map> replicasAssignments, + Map configs) { + this(NO_NUM_PARTITIONS_SIGN, NO_REPLICATION_FACTOR_SIGN, replicasAssignments, configs); + } + + public TopicDetails(Map> replicasAssignments) { + this(replicasAssignments, new HashMap()); + } + } + + private final Map topics; + private final Integer timeout; + + // Set to handle special case where 2 requests for the same topic exist on the wire. + // This allows the broker to return an error code for these topics. + private final Set duplicateTopics; + + public static final int NO_NUM_PARTITIONS_SIGN = -1; + public static final short NO_REPLICATION_FACTOR_SIGN = -1; + + public CreateTopicsRequest(Map topics, Integer timeout) { + super(new Struct(CURRENT_SCHEMA)); + + List createTopicRequestStructs = new ArrayList<>(topics.size()); + for (Map.Entry entry : topics.entrySet()) { + + Struct singleRequestStruct = struct.instance(REQUESTS_KEY_NAME); + String topic = entry.getKey(); + TopicDetails args = entry.getValue(); + + singleRequestStruct.set(TOPIC_KEY_NAME, topic); + singleRequestStruct.set(NUM_PARTITIONS_KEY_NAME, args.numPartitions); + singleRequestStruct.set(REPLICATION_FACTOR_KEY_NAME, args.replicationFactor); + + // replica assignment + List replicaAssignmentsStructs = new ArrayList<>(args.replicasAssignments.size()); + for (Map.Entry> partitionReplicaAssignment : args.replicasAssignments.entrySet()) { + Struct replicaAssignmentStruct = singleRequestStruct.instance(REPLICA_ASSIGNMENT_KEY_NAME); + replicaAssignmentStruct.set(REPLICA_ASSIGNMENT_PARTITION_ID_KEY_NAME, partitionReplicaAssignment.getKey()); + replicaAssignmentStruct.set(REPLICA_ASSIGNMENT_REPLICAS_KEY_NAME, partitionReplicaAssignment.getValue().toArray()); + replicaAssignmentsStructs.add(replicaAssignmentStruct); + } + singleRequestStruct.set(REPLICA_ASSIGNMENT_KEY_NAME, replicaAssignmentsStructs.toArray()); + + // configs + List configsStructs = new ArrayList<>(args.configs.size()); + for (Map.Entry configEntry : args.configs.entrySet()) { + Struct configStruct = singleRequestStruct.instance(CONFIGS_KEY_NAME); + configStruct.set(CONFIG_KEY_KEY_NAME, configEntry.getKey()); + configStruct.set(CONFIG_VALUE_KEY_NAME, configEntry.getValue()); + configsStructs.add(configStruct); + } + singleRequestStruct.set(CONFIGS_KEY_NAME, configsStructs.toArray()); + createTopicRequestStructs.add(singleRequestStruct); + } + struct.set(REQUESTS_KEY_NAME, createTopicRequestStructs.toArray()); + struct.set(TIMEOUT_KEY_NAME, timeout); + + this.topics = topics; + this.timeout = timeout; + this.duplicateTopics = new HashSet<>(); + } + + public CreateTopicsRequest(Struct struct) { + super(struct); + + Object[] requestStructs = struct.getArray(REQUESTS_KEY_NAME); + Map topics = new HashMap<>(); + Set duplicateTopics = new HashSet<>(); + + for (Object requestStructObj : requestStructs) { + Struct singleRequestStruct = (Struct) requestStructObj; + String topic = singleRequestStruct.getString(TOPIC_KEY_NAME); + + if (topics.containsKey(topic)) + duplicateTopics.add(topic); + + int numPartitions = singleRequestStruct.getInt(NUM_PARTITIONS_KEY_NAME); + short replicationFactor = singleRequestStruct.getShort(REPLICATION_FACTOR_KEY_NAME); + + //replica assignment + Object[] assignmentsArray = singleRequestStruct.getArray(REPLICA_ASSIGNMENT_KEY_NAME); + Map> partitionReplicaAssignments = new HashMap<>(assignmentsArray.length); + for (Object assignmentStructObj : assignmentsArray) { + Struct assignmentStruct = (Struct) assignmentStructObj; + + Integer partitionId = assignmentStruct.getInt(REPLICA_ASSIGNMENT_PARTITION_ID_KEY_NAME); + + Object[] replicasArray = assignmentStruct.getArray(REPLICA_ASSIGNMENT_REPLICAS_KEY_NAME); + List replicas = new ArrayList<>(replicasArray.length); + for (Object replica : replicasArray) { + replicas.add((Integer) replica); + } + + partitionReplicaAssignments.put(partitionId, replicas); + } + + Object[] configArray = singleRequestStruct.getArray(CONFIGS_KEY_NAME); + Map configs = new HashMap<>(configArray.length); + for (Object configStructObj : configArray) { + Struct configStruct = (Struct) configStructObj; + + String key = configStruct.getString(CONFIG_KEY_KEY_NAME); + String value = configStruct.getString(CONFIG_VALUE_KEY_NAME); + + configs.put(key, value); + } + + TopicDetails args = new TopicDetails(numPartitions, replicationFactor, partitionReplicaAssignments, configs); + + topics.put(topic, args); + } + + this.topics = topics; + this.timeout = struct.getInt(TIMEOUT_KEY_NAME); + this.duplicateTopics = duplicateTopics; + } + + @Override + public AbstractRequestResponse getErrorResponse(int versionId, Throwable e) { + Map topicErrors = new HashMap<>(); + for (String topic : topics.keySet()) { + topicErrors.put(topic, Errors.forException(e)); + } + + switch (versionId) { + case 0: + return new CreateTopicsResponse(topicErrors); + default: + throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d", + versionId, this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.CREATE_TOPICS.id))); + } + } + + public Map topics() { + return this.topics; + } + + public Integer timeout() { + return this.timeout; + } + + public Set duplicateTopics() { + return this.duplicateTopics; + } + + public static CreateTopicsRequest parse(ByteBuffer buffer, int versionId) { + return new CreateTopicsRequest(ProtoUtils.parseRequest(ApiKeys.CREATE_TOPICS.id, versionId, buffer)); + } + + public static CreateTopicsRequest parse(ByteBuffer buffer) { + return new CreateTopicsRequest(CURRENT_SCHEMA.read(buffer)); + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java new file mode 100644 index 0000000000000..a4fe431a41817 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java @@ -0,0 +1,95 @@ +/** + * 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.requests; + + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.ProtoUtils; +import org.apache.kafka.common.protocol.types.Schema; +import org.apache.kafka.common.protocol.types.Struct; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class CreateTopicsResponse extends AbstractRequestResponse { + private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.CREATE_TOPICS.id); + + private static final String TOPIC_ERROR_CODES_KEY_NAME = "topic_error_codes"; + private static final String TOPIC_KEY_NAME = "topic"; + private static final String ERROR_CODE_KEY_NAME = "error_code"; + + /** + * Possible error codes: + * + * INVALID_TOPIC_EXCEPTION(17) + * CLUSTER_AUTHORIZATION_FAILED(31) + * TOPIC_ALREADY_EXISTS(32) + * INVALID_PARTITIONS(38) + * INVALID_REPLICATION_FACTOR(39) + * INVALID_REPLICA_ASSIGNMENT(40) + * INVALID_CONFIG(41) + * NOT_CONTROLLER(42) + */ + + private final Map errors; + + public CreateTopicsResponse(Map errors) { + super(new Struct(CURRENT_SCHEMA)); + + List topicErrorCodeStructs = new ArrayList<>(errors.size()); + for (Map.Entry topicError : errors.entrySet()) { + Struct topicErrorCodeStruct = struct.instance(TOPIC_ERROR_CODES_KEY_NAME); + topicErrorCodeStruct.set(TOPIC_KEY_NAME, topicError.getKey()); + topicErrorCodeStruct.set(ERROR_CODE_KEY_NAME, topicError.getValue().code()); + topicErrorCodeStructs.add(topicErrorCodeStruct); + } + struct.set(TOPIC_ERROR_CODES_KEY_NAME, topicErrorCodeStructs.toArray()); + + this.errors = errors; + } + + public CreateTopicsResponse(Struct struct) { + super(struct); + + Object[] topicErrorCodesStructs = struct.getArray(TOPIC_ERROR_CODES_KEY_NAME); + Map errors = new HashMap<>(); + for (Object topicErrorCodeStructObj : topicErrorCodesStructs) { + Struct topicErrorCodeStruct = (Struct) topicErrorCodeStructObj; + String topic = topicErrorCodeStruct.getString(TOPIC_KEY_NAME); + short errorCode = topicErrorCodeStruct.getShort(ERROR_CODE_KEY_NAME); + errors.put(topic, Errors.forCode(errorCode)); + } + + this.errors = errors; + } + + public Map errors() { + return errors; + } + + public static CreateTopicsResponse parse(ByteBuffer buffer) { + return new CreateTopicsResponse(CURRENT_SCHEMA.read(buffer)); + } + + public static CreateTopicsResponse parse(ByteBuffer buffer, int version) { + return new CreateTopicsResponse(ProtoUtils.responseSchema(ApiKeys.CREATE_TOPICS.id, version).read(buffer)); + } +} diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index 043582d44b922..2f53a3ce36a45 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -99,7 +99,10 @@ public void testSerialization() throws Exception { createSaslHandshakeResponse(), createApiVersionRequest(), createApiVersionRequest().getErrorResponse(0, new UnknownServerException()), - createApiVersionResponse() + createApiVersionResponse(), + createCreateTopicRequest(), + createCreateTopicRequest().getErrorResponse(0, new UnknownServerException()), + createCreateTopicResponse() ); for (AbstractRequestResponse req : requestResponseList) @@ -451,4 +454,29 @@ private AbstractRequestResponse createApiVersionResponse() { List apiVersions = Arrays.asList(new ApiVersionsResponse.ApiVersion((short) 0, (short) 0, (short) 2)); return new ApiVersionsResponse(Errors.NONE.code(), apiVersions); } + + private AbstractRequest createCreateTopicRequest() { + CreateTopicsRequest.TopicDetails request1 = new CreateTopicsRequest.TopicDetails(3, (short) 5); + + Map> replicaAssignments = new HashMap<>(); + replicaAssignments.put(1, Arrays.asList(1, 2, 3)); + replicaAssignments.put(2, Arrays.asList(2, 3, 4)); + + Map configs = new HashMap<>(); + configs.put("config1", "value1"); + + CreateTopicsRequest.TopicDetails request2 = new CreateTopicsRequest.TopicDetails(replicaAssignments, configs); + + Map request = new HashMap<>(); + request.put("my_t1", request1); + request.put("my_t2", request2); + return new CreateTopicsRequest(request, 0); + } + + private AbstractRequestResponse createCreateTopicResponse() { + Map errors = new HashMap<>(); + errors.put("t1", Errors.INVALID_TOPIC_EXCEPTION); + errors.put("t2", Errors.LEADER_NOT_AVAILABLE); + return new CreateTopicsResponse(errors); + } } diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala index 53b6dd72a0b6b..83c2f6cf44cf8 100644 --- a/core/src/main/scala/kafka/admin/AdminUtils.scala +++ b/core/src/main/scala/kafka/admin/AdminUtils.scala @@ -23,11 +23,10 @@ import kafka.log.LogConfig import kafka.server.ConfigType import kafka.utils._ import kafka.utils.ZkUtils._ - import java.util.Random import java.util.Properties import org.apache.kafka.common.Node -import org.apache.kafka.common.errors.{ReplicaNotAvailableException, InvalidTopicException, LeaderNotAvailableException} +import org.apache.kafka.common.errors.{ReplicaNotAvailableException, InvalidTopicException, LeaderNotAvailableException, InvalidPartitionsException, InvalidReplicationFactorException, TopicExistsException, InvalidReplicaAssignmentException} import org.apache.kafka.common.protocol.{Errors, SecurityProtocol} import org.apache.kafka.common.requests.MetadataResponse @@ -110,11 +109,11 @@ object AdminUtils extends Logging { fixedStartIndex: Int = -1, startPartitionId: Int = -1): Map[Int, Seq[Int]] = { if (nPartitions <= 0) - throw new AdminOperationException("number of partitions must be larger than 0") + throw new InvalidPartitionsException("number of partitions must be larger than 0") if (replicationFactor <= 0) - throw new AdminOperationException("replication factor must be larger than 0") + throw new InvalidReplicationFactorException("replication factor must be larger than 0") if (replicationFactor > brokerMetadatas.size) - throw new AdminOperationException(s"replication factor: $replicationFactor larger than available brokers: ${brokerMetadatas.size}") + throw new InvalidReplicationFactorException(s"replication factor: $replicationFactor larger than available brokers: ${brokerMetadatas.size}") if (brokerMetadatas.forall(_.rack.isEmpty)) assignReplicasToBrokersRackUnaware(nPartitions, replicationFactor, brokerMetadatas.map(_.id), fixedStartIndex, startPartitionId) @@ -411,7 +410,6 @@ object AdminUtils extends Logging { update: Boolean = false) { // validate arguments Topic.validate(topic) - require(partitionReplicaAssignment.values.map(_.size).toSet.size == 1, "All partitions should have the same number of replicas.") val topicPath = getTopicPath(topic) @@ -427,7 +425,14 @@ object AdminUtils extends Logging { } } - partitionReplicaAssignment.values.foreach(reps => require(reps.size == reps.toSet.size, "Duplicate replica assignment found: " + partitionReplicaAssignment)) + if (partitionReplicaAssignment.values.map(_.size).toSet.size != 1) + throw new InvalidReplicaAssignmentException("All partitions should have the same number of replicas") + + partitionReplicaAssignment.values.foreach(reps => + if (reps.size != reps.toSet.size) + throw new InvalidReplicaAssignmentException("Duplicate replica assignment found: " + partitionReplicaAssignment) + ) + // Configs only matter if a topic is being created. Changing configs via AlterTopic is not supported if (!update) { diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 39bfe62ba4705..57a545838cb04 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -19,14 +19,14 @@ package kafka.admin import java.util.Properties import joptsimple._ -import kafka.common.{AdminCommandFailedException, Topic, TopicExistsException} +import kafka.common.{AdminCommandFailedException, Topic} import kafka.consumer.{ConsumerConfig => OldConsumerConfig, Whitelist} -import kafka.coordinator.GroupCoordinator import kafka.log.{Defaults, LogConfig} import kafka.server.ConfigType import kafka.utils.ZkUtils._ import kafka.utils._ import org.I0Itec.zkclient.exception.ZkNodeExistsException +import org.apache.kafka.common.errors.TopicExistsException import org.apache.kafka.common.security.JaasUtils import org.apache.kafka.common.utils.Utils import scala.collection.JavaConversions._ @@ -386,7 +386,7 @@ object TopicCommand extends Logging { "*****************************************************************************************************\n" + "*** WARNING: you are creating a topic where the max.message.bytes is greater than the broker's ***\n" + "*** default max.message.bytes. This operation is potentially dangerous. Consumers will get ***\n" + - s"*** failures if their fetch.message.max.bytes (old consumer) or ${NewConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG} ***\n"+ + s"*** failures if their fetch.message.max.bytes (old consumer) or ${NewConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG} ***\n"+ "*** (new consumer) < the value you are using. ***\n" + "*****************************************************************************************************\n" + s"- value set here: $maxMessageBytes\n" + diff --git a/core/src/main/scala/kafka/cluster/Broker.scala b/core/src/main/scala/kafka/cluster/Broker.scala index 61290c1fe6730..7116722ab4fff 100755 --- a/core/src/main/scala/kafka/cluster/Broker.scala +++ b/core/src/main/scala/kafka/cluster/Broker.scala @@ -17,9 +17,6 @@ package kafka.cluster -import java.nio.ByteBuffer - -import kafka.api.ApiUtils._ import kafka.common.{BrokerEndPointNotAvailableException, BrokerNotAvailableException, KafkaException} import kafka.utils.Json import org.apache.kafka.common.Node diff --git a/core/src/main/scala/kafka/common/ErrorMapping.scala b/core/src/main/scala/kafka/common/ErrorMapping.scala index 91a1d757559c8..bbbc854c8d4fc 100644 --- a/core/src/main/scala/kafka/common/ErrorMapping.scala +++ b/core/src/main/scala/kafka/common/ErrorMapping.scala @@ -20,6 +20,7 @@ package kafka.common import java.nio.ByteBuffer import kafka.message.InvalidMessageException +import org.apache.kafka.common.errors.InvalidTopicException import scala.Predef._ diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index 1584cc95a38ee..7548cc1957314 100755 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/core/src/main/scala/kafka/controller/KafkaController.scala @@ -180,6 +180,8 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat private val preferredReplicaElectionListener = new PreferredReplicaElectionListener(this) private val isrChangeNotificationListener = new IsrChangeNotificationListener(this) + val topicPurgatory = DelayedOperationPurgatory[DelayedOperation]("topic", config.brokerId) + newGauge( "ActiveControllerCount", new Gauge[Int] { @@ -225,6 +227,15 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat "id_%d-host_%s-port_%d".format(config.brokerId, controllerListener.get.host, controllerListener.get.port) } + /** + * Try to complete delayed topic operations with the request key + */ + def tryCompleteDelayedTopicOperations(topic: String) { + val key = TopicKey(topic) + val completed = topicPurgatory.checkAndComplete(key) + debug(s"Request key ${key.keyLabel} unblocked $completed topic requests.") + } + /** * On clean shutdown, the controller first determines the partitions that the * shutting down broker leads, and moves leadership of those partitions to another broker @@ -506,6 +517,7 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat // subscribe to partition changes topics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic)) onNewPartitionCreation(newPartitions) + topics.foreach(topic => tryCompleteDelayedTopicOperations(topic)) } /** diff --git a/core/src/main/scala/kafka/log/LogConfig.scala b/core/src/main/scala/kafka/log/LogConfig.scala index d5e06fa13e6c8..31e62b4e01d2b 100755 --- a/core/src/main/scala/kafka/log/LogConfig.scala +++ b/core/src/main/scala/kafka/log/LogConfig.scala @@ -24,6 +24,7 @@ import scala.collection.JavaConverters._ import kafka.api.ApiVersion import kafka.message.{BrokerCompressionCodec, Message} import kafka.server.KafkaConfig +import org.apache.kafka.common.errors.InvalidConfigurationException import org.apache.kafka.common.config.{AbstractConfig, ConfigDef} import org.apache.kafka.common.record.TimestampType import org.apache.kafka.common.utils.Utils @@ -181,7 +182,7 @@ object LogConfig { def apply(): LogConfig = LogConfig(new Properties()) def configNames: Seq[String] = configDef.names.asScala.toSeq.sorted - + /** * Create a log config instance using the given properties and defaults */ @@ -197,7 +198,9 @@ object LogConfig { */ def validateNames(props: Properties) { val names = configNames - for (name <- props.keys.asScala) require(names.contains(name), s"Unknown configuration `$name`.") + for(name <- props.asScala.keys) + if (!names.contains(name)) + throw new InvalidConfigurationException(s"Unknown configuration $name.") } /** diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala index 53a2346b42a30..cff7b1a9537f0 100644 --- a/core/src/main/scala/kafka/network/RequestChannel.scala +++ b/core/src/main/scala/kafka/network/RequestChannel.scala @@ -27,9 +27,10 @@ import kafka.api._ import kafka.metrics.KafkaMetricsGroup import kafka.utils.{Logging, SystemTime} import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.errors.InvalidRequestException import org.apache.kafka.common.network.Send -import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol, Protocol} -import org.apache.kafka.common.requests.{RequestSend, ProduceRequest, AbstractRequest, RequestHeader, ApiVersionsRequest} +import org.apache.kafka.common.protocol.{ApiKeys, Protocol, SecurityProtocol} +import org.apache.kafka.common.requests.{AbstractRequest, ApiVersionsRequest, ProduceRequest, RequestHeader, RequestSend} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.log4j.Logger diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index f00dd7b361253..f9ce58a0b659e 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -31,6 +31,7 @@ import kafka.common.KafkaException import kafka.metrics.KafkaMetricsGroup import kafka.server.KafkaConfig import kafka.utils._ +import org.apache.kafka.common.errors.InvalidRequestException import org.apache.kafka.common.metrics._ import org.apache.kafka.common.network.{ChannelBuilders, KafkaChannel, LoginType, Mode, Selectable, Selector => KSelector} import org.apache.kafka.common.security.auth.KafkaPrincipal diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala new file mode 100644 index 0000000000000..bee8c48df44d1 --- /dev/null +++ b/core/src/main/scala/kafka/server/AdminManager.scala @@ -0,0 +1,103 @@ +/** + * 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 kafka.server + +import java.util.Properties + +import kafka.admin.AdminUtils +import kafka.controller.KafkaController +import kafka.log.LogConfig +import kafka.metrics.KafkaMetricsGroup +import kafka.utils._ +import org.apache.kafka.common.errors.InvalidRequestException +import org.apache.kafka.common.metrics.Metrics +import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.requests.CreateTopicsRequest._ + +import scala.collection._ +import scala.collection.JavaConverters._ + +class AdminManager(val config: KafkaConfig, + val metrics: Metrics, + val controller: KafkaController, + val metadataCache: MetadataCache, + val zkUtils: ZkUtils) extends Logging with KafkaMetricsGroup { + this.logIdent = "[Admin Manager on Broker " + config.brokerId + "]: " + + /** + * Create topics and wait until the topics have been completely created. + * The callback function will be triggered either when timeout, error or the topics are created. + */ + def createTopics(timeout: Int, + createInfo: Map[String, TopicDetails], + responseCallback: Map[String, Errors] => Unit) { + + // 1. map over topics creating assignment and calling zookeeper + val brokers = metadataCache.getAliveBrokers.map { b => kafka.admin.BrokerMetadata(b.id, b.rack) } + val metadata = createInfo.map { case (topic, arguments) => + try { + val configs = new Properties() + arguments.configs.asScala.foreach { case (key, value) => + configs.setProperty(key, value) + } + LogConfig.validate(configs) + + val assignments = { + if ((arguments.numPartitions != NO_NUM_PARTITIONS_SIGN || arguments.replicationFactor != NO_REPLICATION_FACTOR_SIGN) + && !arguments.replicasAssignments.isEmpty) + throw new InvalidRequestException("Both numPartitions or replicationFactor and replicasAssignments were set. " + + "Both cannot be used at the same time.") + else if (!arguments.replicasAssignments.isEmpty) { + // Note: we don't check that replicaAssignment doesn't contain unknown brokers - unlike in add-partitions case, + // this follows the existing logic in TopicCommand + arguments.replicasAssignments.asScala.map { case (partitionId, replicas) => + (partitionId.intValue, replicas.asScala.map(_.intValue)) + } + } else { + AdminUtils.assignReplicasToBrokers(brokers, arguments.numPartitions, arguments.replicationFactor) + } + } + trace(s"Assignments for topic $topic are $assignments ") + AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topic, assignments, configs, update = false) + CreateTopicMetadata(topic, assignments, Errors.NONE) + } catch { + case e: Throwable => + error(s"Error processing create topic request for topic $topic with arguments $arguments", e) + CreateTopicMetadata(topic, Map(), Errors.forException(e)) + } + } + + // 2. if timeout <= 0 or no topics can proceed return immediately + if (timeout <= 0 || !metadata.exists(_.error == Errors.NONE)) { + val results = metadata.map { createTopicMetadata => + // ignore topics that already have errors + if (createTopicMetadata.error == Errors.NONE) { + (createTopicMetadata.topic, Errors.REQUEST_TIMED_OUT) + } else { + (createTopicMetadata.topic, createTopicMetadata.error) + } + }.toMap + responseCallback(results) + } else { + // 3. else pass the assignments and errors to the delayed operation and set the keys + val delayedCreate = new DelayedCreateTopics(timeout, metadata.toSeq, this, responseCallback) + val delayedCreateKeys = createInfo.keys.map(new TopicKey(_)).toSeq + // try to complete the request immediately, otherwise put it into the purgatory + controller.topicPurgatory.tryCompleteElseWatch(delayedCreate, delayedCreateKeys) + } + } +} diff --git a/core/src/main/scala/kafka/server/DelayedCreateTopics.scala b/core/src/main/scala/kafka/server/DelayedCreateTopics.scala new file mode 100644 index 0000000000000..720e109cc68e1 --- /dev/null +++ b/core/src/main/scala/kafka/server/DelayedCreateTopics.scala @@ -0,0 +1,94 @@ +/** + * 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 kafka.server + +import kafka.api.LeaderAndIsr +import kafka.common.TopicAndPartition +import org.apache.kafka.common.protocol.Errors + +import scala.collection._ + +/** + * The create metadata maintained by the delayed create operation + * + * TODO: local state doesn't count, need to know state of all relevant brokers + * + */ +case class CreateTopicMetadata(topic: String, replicaAssignments: Map[Int, Seq[Int]], error: Errors) + +/** + * A delayed create topics operation that can be created by the admin manager and watched + * in the topic purgatory + */ +class DelayedCreateTopics(delayMs: Long, + createMetadata: Seq[CreateTopicMetadata], + adminManager: AdminManager, + responseCallback: Map[String, Errors] => Unit) + extends DelayedOperation(delayMs) { + + /** + * The operation can be completed if all of the topics exist every partition has a leader in the controller. + * See KafkaController.onNewTopicCreation + */ + override def tryComplete() : Boolean = { + trace(s"Trying to complete operation for $createMetadata") + trace(s"Live brokers ${adminManager.controller.controllerContext.liveBrokerIds}") + + val leaderlessPartitionCount = createMetadata.foldLeft(0) { case (topicCounter, metadata) => + topicCounter + missingLeaderCount(metadata.topic, metadata.replicaAssignments.keySet) + } + + if (leaderlessPartitionCount == 0) { + trace("All partitions have a leader, completing the delayed operation") + forceComplete() + } else { + trace(s"$leaderlessPartitionCount partitions do not have a leader, not completing the delayed operation") + false + } + } + + /** + * Check for partitions that are still missing a leader, update their error code and call the responseCallback + */ + override def onComplete() { + trace(s"Completing operation for $createMetadata") + val results = createMetadata.map { metadata => + // ignore topics that already have errors + if (metadata.error == Errors.NONE && missingLeaderCount(metadata.topic, metadata.replicaAssignments.keySet) > 0) + (metadata.topic, Errors.REQUEST_TIMED_OUT) + else + (metadata.topic, metadata.error) + }.toMap + responseCallback(results) + } + + override def onExpiration(): Unit = { } + + private def missingLeaderCount(topic: String, partitions: Set[Int]): Int = { + partitions.foldLeft(0) { case (counter, partition) => + if (isMissingLeader(topic, partition)) counter + 1 else counter + } + } + + private def isMissingLeader(topic: String, partition: Int): Boolean = { + val leadershipInfo = adminManager.controller.controllerContext.partitionLeadershipInfo + val leaderInfo =leadershipInfo.get(TopicAndPartition(topic, partition)) + val partitionInfo = adminManager.metadataCache.getPartitionInfo(topic, partition) + leaderInfo.isEmpty || leaderInfo.get.leaderAndIsr.leader == LeaderAndIsr.NoLeader + } +} diff --git a/core/src/main/scala/kafka/server/DelayedOperationKey.scala b/core/src/main/scala/kafka/server/DelayedOperationKey.scala index 072a658ea25b7..0e05cce1f6ff2 100644 --- a/core/src/main/scala/kafka/server/DelayedOperationKey.scala +++ b/core/src/main/scala/kafka/server/DelayedOperationKey.scala @@ -52,3 +52,9 @@ case class GroupKey(groupId: String) extends DelayedOperationKey { override def keyLabel = groupId } + +/* used by delayed-topic operations */ +case class TopicKey(topic: String) extends DelayedOperationKey { + + override def keyLabel = topic +} diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index ebd173226c396..72bc7cd54460b 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -34,10 +34,10 @@ import kafka.network._ import kafka.network.RequestChannel.{Response, Session} import kafka.security.auth.{Authorizer, ClusterAction, Create, Describe, Group, Operation, Read, Resource, Topic, Write} import kafka.utils.{Logging, SystemTime, ZKGroupTopicDirs, ZkUtils} -import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidTopicException, NotLeaderForPartitionException, UnknownTopicOrPartitionException} +import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidTopicException, NotLeaderForPartitionException, UnknownTopicOrPartitionException, TopicExistsException} import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.protocol.{ApiKeys, Errors, Protocol, SecurityProtocol} -import org.apache.kafka.common.requests.{ApiVersionsResponse, DescribeGroupsRequest, DescribeGroupsResponse, GroupCoordinatorRequest, GroupCoordinatorResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaderAndIsrRequest, LeaderAndIsrResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsResponse, ListOffsetRequest, ListOffsetResponse, MetadataRequest, MetadataResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetFetchRequest, OffsetFetchResponse, ProduceRequest, ProduceResponse, ResponseHeader, ResponseSend, StopReplicaRequest, StopReplicaResponse, SyncGroupRequest, SyncGroupResponse, UpdateMetadataRequest, UpdateMetadataResponse} +import org.apache.kafka.common.requests.{ApiVersionsResponse, DescribeGroupsRequest, DescribeGroupsResponse, GroupCoordinatorRequest, GroupCoordinatorResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaderAndIsrRequest, LeaderAndIsrResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsResponse, ListOffsetRequest, ListOffsetResponse, MetadataRequest, MetadataResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetFetchRequest, OffsetFetchResponse, ProduceRequest, ProduceResponse, ResponseHeader, ResponseSend, StopReplicaRequest, StopReplicaResponse, SyncGroupRequest, SyncGroupResponse, UpdateMetadataRequest, UpdateMetadataResponse, CreateTopicsRequest, CreateTopicsResponse} import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.utils.Utils import org.apache.kafka.common.{Node, TopicPartition} @@ -52,6 +52,7 @@ import org.apache.kafka.common.requests.SaslHandshakeResponse */ class KafkaApis(val requestChannel: RequestChannel, val replicaManager: ReplicaManager, + val adminManager: AdminManager, val coordinator: GroupCoordinator, val controller: KafkaController, val zkUtils: ZkUtils, @@ -92,6 +93,7 @@ class KafkaApis(val requestChannel: RequestChannel, case ApiKeys.LIST_GROUPS => handleListGroupsRequest(request) case ApiKeys.SASL_HANDSHAKE => handleSaslHandshakeRequest(request) case ApiKeys.API_VERSIONS => handleApiVersionsRequest(request) + case ApiKeys.CREATE_TOPICS => handleCreateTopicsRequest(request) case requestId => throw new KafkaException("Unknown api code " + requestId) } } catch { @@ -1043,6 +1045,43 @@ class KafkaApis(val requestChannel: RequestChannel, info("Shutdown complete.") } + def handleCreateTopicsRequest(request: RequestChannel.Request) { + val createTopicsRequest = request.body.asInstanceOf[CreateTopicsRequest] + + val (validTopics, duplicateTopics) = createTopicsRequest.topics.asScala.partition { case (topic, _) => + !createTopicsRequest.duplicateTopics.contains(topic) + } + + def sendResponseCallback(results: Map[String, Errors]): Unit = { + val respHeader = new ResponseHeader(request.header.correlationId) + if(duplicateTopics.nonEmpty) + error("Multiple entries for the same topic were found for the following topics: " + duplicateTopics.keySet.mkString(",")) + val completeResults = results ++ duplicateTopics.keySet.map((_, Errors.INVALID_REQUEST)).toMap + val responseBody = new CreateTopicsResponse(completeResults.asJava) + trace(s"Sending create topics response $responseBody for correlation id ${request.header.correlationId} to client ${request.header.clientId}.") + requestChannel.sendResponse(new RequestChannel.Response(request, new ResponseSend(request.connectionId, respHeader, responseBody))) + } + + if(!controller.isActive()) { + val results = createTopicsRequest.topics.asScala.map { case (topic, _) => + (topic, Errors.NOT_CONTROLLER) + } + sendResponseCallback(results) + } else if (!authorize(request.session, Create, Resource.ClusterResource)) { + val results = createTopicsRequest.topics.asScala.map { case (topic, _) => + (topic, Errors.CLUSTER_AUTHORIZATION_FAILED) + } + sendResponseCallback(results) + } + else { + adminManager.createTopics( + createTopicsRequest.timeout.toInt, + validTopics, + sendResponseCallback + ) + } + } + def authorizeClusterAction(request: RequestChannel.Request): Unit = { if (!authorize(request.session, ClusterAction, Resource.ClusterResource)) throw new ClusterAuthorizationException(s"Request $request is not authorized.") diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 994e28e647629..07f8dd1a02512 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -119,6 +119,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr var logManager: LogManager = null var replicaManager: ReplicaManager = null + var adminManager: AdminManager = null var dynamicConfigHandlers: Map[String, ConfigHandler] = null var dynamicConfigManager: DynamicConfigManager = null @@ -199,6 +200,9 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr kafkaController = new KafkaController(config, zkUtils, brokerState, kafkaMetricsTime, metrics, threadNamePrefix) kafkaController.startup() + /* start admin manager */ + adminManager = new AdminManager(config, metrics, kafkaController, metadataCache, zkUtils) + /* start group coordinator */ groupCoordinator = GroupCoordinator(config, zkUtils, replicaManager, kafkaMetricsTime) groupCoordinator.startup() @@ -211,7 +215,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr } /* start processing requests */ - apis = new KafkaApis(socketServer.requestChannel, replicaManager, groupCoordinator, + apis = new KafkaApis(socketServer.requestChannel, replicaManager, adminManager, groupCoordinator, kafkaController, zkUtils, config.brokerId, config, metadataCache, metrics, authorizer) requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads) diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index 10e0baed74755..b1f0283a62142 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -20,7 +20,6 @@ import java.util.{ArrayList, Collections, Properties} import kafka.cluster.EndPoint import kafka.common.TopicAndPartition -import kafka.coordinator.GroupCoordinator import kafka.integration.KafkaServerTestHarness import kafka.security.auth._ import kafka.server.KafkaConfig @@ -30,6 +29,7 @@ import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord} import org.apache.kafka.common.errors._ import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol} import org.apache.kafka.common.requests._ +import CreateTopicsRequest.TopicDetails import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.{Node, TopicPartition, requests} import org.junit.Assert._ @@ -42,6 +42,7 @@ import org.apache.kafka.common.internals.TopicConstants class AuthorizerIntegrationTest extends KafkaServerTestHarness { val topic = "topic" + val createTopic = "topic-new" val part = 0 val brokerId: Integer = 0 val correlationId = 0 @@ -54,6 +55,7 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness { val GroupReadAcl = Map(groupResource -> Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read))) val ClusterAcl = Map(Resource.ClusterResource -> Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, ClusterAction))) + val ClusterCreateAcl = Map(Resource.ClusterResource -> Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Create))) val TopicReadAcl = Map(topicResource -> Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read))) val TopicWriteAcl = Map(topicResource -> Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write))) val TopicDescribeAcl = Map(topicResource -> Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Describe))) @@ -89,8 +91,9 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness { ApiKeys.LEAVE_GROUP.id -> classOf[LeaveGroupResponse], ApiKeys.LEADER_AND_ISR.id -> classOf[requests.LeaderAndIsrResponse], ApiKeys.STOP_REPLICA.id -> classOf[requests.StopReplicaResponse], - ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> classOf[requests.ControlledShutdownResponse] - ) + ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> classOf[requests.ControlledShutdownResponse], + ApiKeys.CREATE_TOPICS.id -> classOf[CreateTopicsResponse] + ) val RequestKeyToErrorCode = Map[Short, (Nothing) => Short]( ApiKeys.METADATA.id -> ((resp: requests.MetadataResponse) => resp.errors().asScala.find(_._1 == topic).getOrElse(("test", Errors.NONE))._2.code()), @@ -107,7 +110,8 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness { ApiKeys.LEAVE_GROUP.id -> ((resp: LeaveGroupResponse) => resp.errorCode()), ApiKeys.LEADER_AND_ISR.id -> ((resp: requests.LeaderAndIsrResponse) => resp.responses().asScala.find(_._1 == tp).get._2), ApiKeys.STOP_REPLICA.id -> ((resp: requests.StopReplicaResponse) => resp.responses().asScala.find(_._1 == tp).get._2), - ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> ((resp: requests.ControlledShutdownResponse) => resp.errorCode()) + ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> ((resp: requests.ControlledShutdownResponse) => resp.errorCode()), + ApiKeys.CREATE_TOPICS.id -> ((resp: CreateTopicsResponse) => resp.errors().asScala.find(_._1 == createTopic).get._2.code) ) val RequestKeysToAcls = Map[Short, Map[Resource, Set[Acl]]]( @@ -125,7 +129,8 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness { ApiKeys.LEAVE_GROUP.id -> GroupReadAcl, ApiKeys.LEADER_AND_ISR.id -> ClusterAcl, ApiKeys.STOP_REPLICA.id -> ClusterAcl, - ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> ClusterAcl + ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> ClusterAcl, + ApiKeys.CREATE_TOPICS.id -> ClusterCreateAcl ) // configure the servers and clients @@ -227,6 +232,10 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness { new requests.ControlledShutdownRequest(brokerId) } + private def createTopicsRequest = { + new CreateTopicsRequest(Map(createTopic -> new TopicDetails(1, 1.toShort)).asJava, 0) + } + @Test def testAuthorization() { val requestKeyToRequest = mutable.LinkedHashMap[Short, AbstractRequest]( @@ -244,7 +253,8 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness { ApiKeys.LEAVE_GROUP.id -> createLeaveGroupRequest, ApiKeys.LEADER_AND_ISR.id -> createLeaderAndIsrRequest, ApiKeys.STOP_REPLICA.id -> createStopReplicaRequest, - ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> createControlledShutdownRequest + ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> createControlledShutdownRequest, + ApiKeys.CREATE_TOPICS.id -> createTopicsRequest ) val socket = new Socket("localhost", servers.head.boundPort()) diff --git a/core/src/test/scala/unit/kafka/admin/AdminTest.scala b/core/src/test/scala/unit/kafka/admin/AdminTest.scala index 7df141143fded..238cbadeea289 100755 --- a/core/src/test/scala/unit/kafka/admin/AdminTest.scala +++ b/core/src/test/scala/unit/kafka/admin/AdminTest.scala @@ -16,7 +16,7 @@ */ package kafka.admin -import org.apache.kafka.common.errors.InvalidTopicException +import org.apache.kafka.common.errors.{InvalidReplicaAssignmentException, InvalidReplicationFactorException, InvalidTopicException, TopicExistsException} import org.apache.kafka.common.metrics.Quota import org.apache.kafka.common.protocol.ApiKeys import org.junit.Assert._ @@ -27,7 +27,7 @@ import kafka.utils._ import kafka.log._ import kafka.zk.ZooKeeperTestHarness import kafka.utils.{Logging, TestUtils, ZkUtils} -import kafka.common.{TopicAndPartition, TopicExistsException} +import kafka.common.TopicAndPartition import kafka.server.{ConfigType, KafkaConfig, KafkaServer} import java.io.File @@ -42,12 +42,12 @@ class AdminTest extends ZooKeeperTestHarness with Logging with RackAwareTest { val brokerMetadatas = (0 to 4).map(new BrokerMetadata(_, None)) // test 0 replication factor - intercept[AdminOperationException] { + intercept[InvalidReplicationFactorException] { AdminUtils.assignReplicasToBrokers(brokerMetadatas, 10, 0) } // test wrong replication factor - intercept[AdminOperationException] { + intercept[InvalidReplicationFactorException] { AdminUtils.assignReplicasToBrokers(brokerMetadatas, 10, 6) } @@ -74,12 +74,12 @@ class AdminTest extends ZooKeeperTestHarness with Logging with RackAwareTest { TestUtils.createBrokersInZk(zkUtils, brokers) // duplicate brokers - intercept[IllegalArgumentException] { + intercept[InvalidReplicaAssignmentException] { AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, "test", Map(0->Seq(0,0))) } // inconsistent replication factor - intercept[IllegalArgumentException] { + intercept[InvalidReplicaAssignmentException] { AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, "test", Map(0->Seq(0,1), 1->Seq(0))) } diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala index e0107da4f3c37..11dc36eb36416 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -16,7 +16,6 @@ */ package kafka.admin -import kafka.common.TopicExistsException import org.junit.Assert._ import org.junit.Test import kafka.utils.Logging @@ -25,8 +24,9 @@ import kafka.zk.ZooKeeperTestHarness import kafka.server.ConfigType import kafka.admin.TopicCommand.TopicCommandOptions import kafka.utils.ZkUtils._ -import kafka.coordinator.GroupCoordinator import org.apache.kafka.common.internals.TopicConstants +import org.apache.kafka.common.errors.TopicExistsException + class TopicCommandTest extends ZooKeeperTestHarness with Logging with RackAwareTest { diff --git a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala index 906c4b27b2e4a..78c8f6e57f7cd 100644 --- a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala @@ -29,6 +29,8 @@ import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol} import org.apache.kafka.common.requests.{AbstractRequest, RequestHeader, ResponseHeader} import org.junit.Before +import scala.collection.mutable.Buffer + abstract class BaseRequestTest extends KafkaServerTestHarness { private var correlationId = 0 @@ -52,14 +54,32 @@ abstract class BaseRequestTest extends KafkaServerTestHarness { TestUtils.waitUntilTrue(() => servers.head.metadataCache.getAliveBrokers.size == numBrokers, "Wait for cache to update") } - def socketServer = { + def anySocketServer = { servers.find { server => val state = server.brokerState.currentState state != NotRunning.state && state != BrokerShuttingDown.state }.map(_.socketServer).getOrElse(throw new IllegalStateException("No live broker is available")) } - def connect(s: SocketServer = socketServer, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): Socket = { + def controllerSocketServer = { + servers.find { server => + server.kafkaController.isActive() + }.map(_.socketServer).getOrElse(throw new IllegalStateException("No controller broker is available")) + } + + def notControllerSocketServer = { + servers.find { server => + !server.kafkaController.isActive() + }.map(_.socketServer).getOrElse(throw new IllegalStateException("No non-controller broker is available")) + } + + def brokerSocketServer(brokerId: Int) = { + servers.find { server => + server.config.brokerId == brokerId + }.map(_.socketServer).getOrElse(throw new IllegalStateException(s"Could not find broker with id $brokerId")) + } + + def connect(s: SocketServer = anySocketServer, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): Socket = { new Socket("localhost", s.boundPort(protocol)) } @@ -83,19 +103,21 @@ abstract class BaseRequestTest extends KafkaServerTestHarness { receiveResponse(socket) } - def send(request: AbstractRequest, apiKey: ApiKeys, version: Short): ByteBuffer = { - val socket = connect() + def send(request: AbstractRequest, apiKey: ApiKeys, version: Short, + destination: SocketServer = anySocketServer, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): ByteBuffer = { + val socket = connect(destination, protocol) try { - send(socket, request, apiKey, version) + send(request, apiKey, version, socket) } finally { socket.close() } } /** - * Serializes and send the request to the given api. A ByteBuffer containing the response is returned. + * Serializes and send the request to the given api. + * A ByteBuffer containing the response is returned. */ - def send(socket: Socket, request: AbstractRequest, apiKey: ApiKeys, version: Short): ByteBuffer = { + def send(request: AbstractRequest, apiKey: ApiKeys, version: Short, socket: Socket): ByteBuffer = { correlationId += 1 val serializedBytes = { val header = new RequestHeader(apiKey.id, version, "", correlationId) diff --git a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala new file mode 100644 index 0000000000000..6f815412411d9 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala @@ -0,0 +1,212 @@ +/** + * 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 kafka.server + +import kafka.network.SocketServer +import kafka.utils._ +import org.apache.kafka.common.protocol.types.Struct +import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils} +import org.apache.kafka.common.requests.{CreateTopicsRequest, CreateTopicsResponse, MetadataRequest, MetadataResponse} +import org.junit.Assert._ +import org.junit.Test + +import scala.collection.JavaConverters._ + + +class CreateTopicsRequestTest extends BaseRequestTest { + + @Test + def testValidCreateTopicsRequests() { + val timeout = 10000 + // Generated assignments + validateValidCreateTopicsRequests(new CreateTopicsRequest(Map("topic1" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, timeout)) + validateValidCreateTopicsRequests(new CreateTopicsRequest(Map("topic2" -> new CreateTopicsRequest.TopicDetails(1, 3.toShort)).asJava, timeout)) + val config3 = Map("min.insync.replicas" -> "2").asJava + validateValidCreateTopicsRequests(new CreateTopicsRequest(Map("topic3" -> new CreateTopicsRequest.TopicDetails(5, 2.toShort, config3)).asJava, timeout)) + // Manual assignments + val assignments4 = replicaAssignmentToJava(Map(0 -> List(0))) + validateValidCreateTopicsRequests(new CreateTopicsRequest(Map("topic4" -> new CreateTopicsRequest.TopicDetails(assignments4)).asJava, timeout)) + val assignments5 = replicaAssignmentToJava(Map(0 -> List(0, 1), 1 -> List(1, 0), 2 -> List(1, 2))) + val config5 = Map("min.insync.replicas" -> "2").asJava + validateValidCreateTopicsRequests(new CreateTopicsRequest(Map("topic5" -> new CreateTopicsRequest.TopicDetails(assignments5, config5)).asJava, timeout)) + // Mixed + val assignments8 = replicaAssignmentToJava(Map(0 -> List(0, 1), 1 -> List(1, 0), 2 -> List(1, 2))) + validateValidCreateTopicsRequests(new CreateTopicsRequest(Map( + "topic6" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort), + "topic7" -> new CreateTopicsRequest.TopicDetails(5, 2.toShort), + "topic8" -> new CreateTopicsRequest.TopicDetails(assignments8)).asJava, timeout) + ) + } + + private def validateValidCreateTopicsRequests(request: CreateTopicsRequest): Unit = { + val response = sendCreateTopicRequest(request, 0) + + val error = response.errors.values.asScala.find(_ != Errors.NONE) + assertTrue(s"There should be no errors, found ${response.errors.asScala}", error.isEmpty) + + request.topics.asScala.foreach { case (topic, details) => + TestUtils.waitUntilMetadataIsPropagated(servers, topic, 0) + val metadata = sendMetadataRequest(new MetadataRequest(List(topic).asJava)).topicMetadata.asScala + val metadataForTopic = metadata.filter(p => p.topic.equals(topic)).head + + val partitions = if (!details.replicasAssignments.isEmpty) + details.replicasAssignments.size + else + details.numPartitions + + val replication = if (!details.replicasAssignments.isEmpty) + details.replicasAssignments.asScala.head._2.size + else + details.replicationFactor + + assertNotNull("The topic should be created", metadataForTopic) + assertEquals("The topic should have the correct number of partitions", partitions, metadataForTopic.partitionMetadata.size) + assertEquals("The topic should have the correct replication factor", replication, metadataForTopic.partitionMetadata.asScala.head.replicas.size) + } + } + + @Test + def testErrorCreateTopicsRequests() { + val timeout = 10000 + val existingTopic = "existing-topic" + TestUtils.createTopic(zkUtils, existingTopic, 1, 1, servers) + + // Basic + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map(existingTopic -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, timeout), + Map(existingTopic -> Errors.TOPIC_ALREADY_EXISTS)) + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-partitions" -> new CreateTopicsRequest.TopicDetails(-1, 1.toShort)).asJava, timeout), + Map("error-partitions" -> Errors.INVALID_PARTITIONS)) + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-replication" -> new CreateTopicsRequest.TopicDetails(1, (numBrokers + 1).toShort)).asJava, timeout), + Map("error-replication" -> Errors.INVALID_REPLICATION_FACTOR)) + val invalidConfig = Map("not.a.property" -> "error").asJava + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-config" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort, invalidConfig)).asJava, timeout), + Map("error-config" -> Errors.INVALID_CONFIG)) + val invalidAssignments = replicaAssignmentToJava(Map(0 -> List(0, 1), 1 -> List(0))) + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-assignment" -> new CreateTopicsRequest.TopicDetails(invalidAssignments)).asJava, timeout), + Map("error-assignment" -> Errors.INVALID_REPLICA_ASSIGNMENT)) + + // Partial + validateErrorCreateTopicsRequests( + new CreateTopicsRequest(Map( + existingTopic -> new CreateTopicsRequest.TopicDetails(1, 1.toShort), + "partial-partitions" -> new CreateTopicsRequest.TopicDetails(-1, 1.toShort), + "partial-replication" -> new CreateTopicsRequest.TopicDetails(1, (numBrokers + 1).toShort), + "partial-none" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, timeout), + Map( + existingTopic -> Errors.TOPIC_ALREADY_EXISTS, + "partial-partitions" -> Errors.INVALID_PARTITIONS, + "partial-replication" -> Errors.INVALID_REPLICATION_FACTOR, + "partial-none" -> Errors.NONE + ) + ) + validateTopicExists("partial-none") + + // Timeout + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-timeout" -> new CreateTopicsRequest.TopicDetails(10, 3.toShort)).asJava, 1), + Map("error-timeout" -> Errors.REQUEST_TIMED_OUT)) + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-timeout-zero" -> new CreateTopicsRequest.TopicDetails(10, 3.toShort)).asJava, 0), + Map("error-timeout-zero" -> Errors.REQUEST_TIMED_OUT)) + // Negative timeouts are treated the same as 0 + validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-timeout-negative" -> new CreateTopicsRequest.TopicDetails(10, 3.toShort)).asJava, -1), + Map("error-timeout-negative" -> Errors.REQUEST_TIMED_OUT)) + // The topics should still get created eventually + TestUtils.waitUntilMetadataIsPropagated(servers, "error-timeout", 0) + TestUtils.waitUntilMetadataIsPropagated(servers, "error-timeout-zero", 0) + TestUtils.waitUntilMetadataIsPropagated(servers, "error-timeout-negative", 0) + validateTopicExists("error-timeout") + validateTopicExists("error-timeout-zero") + validateTopicExists("error-timeout-negative") + } + + @Test + def testInvalidCreateTopicsRequests() { + // Duplicate + val singleRequest = new CreateTopicsRequest(Map("duplicate-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, 1000) + val duplicateRequest = duplicateFirstTopic(singleRequest) + assertFalse("Request doesn't have duplicate topics", duplicateRequest.duplicateTopics().isEmpty) + validateErrorCreateTopicsRequests(duplicateRequest, Map("duplicate-topic" -> Errors.INVALID_REQUEST)) + + // Partitions/ReplicationFactor and ReplicaAssignment + val assignments = replicaAssignmentToJava(Map(0 -> List(0))) + val assignmentRequest = new CreateTopicsRequest(Map("bad-args-topic" -> new CreateTopicsRequest.TopicDetails(assignments)).asJava, 1000) + val badArgumentsRequest = addPartitionsAndReplicationFactorToFirstTopic(assignmentRequest) + validateErrorCreateTopicsRequests(badArgumentsRequest, Map("bad-args-topic" -> Errors.INVALID_REQUEST)) + } + + private def duplicateFirstTopic(request: CreateTopicsRequest) = { + val struct = request.toStruct + val topics = struct.getArray("create_topic_requests") + val firstTopic= topics(0).asInstanceOf[Struct] + val newTopics = firstTopic :: topics.toList + struct.set("create_topic_requests", newTopics.toArray) + new CreateTopicsRequest(struct) + } + + private def addPartitionsAndReplicationFactorToFirstTopic(request: CreateTopicsRequest) = { + val struct = request.toStruct + val topics = struct.getArray("create_topic_requests") + val firstTopic = topics(0).asInstanceOf[Struct] + firstTopic.set("num_partitions", 1) + firstTopic.set("replication_factor", 1.toShort) + new CreateTopicsRequest(struct) + } + + private def validateErrorCreateTopicsRequests(request: CreateTopicsRequest, expectedResponse: Map[String, Errors]): Unit = { + val response = sendCreateTopicRequest(request, 0) + val errors = response.errors.asScala + assertEquals("The response size should match", expectedResponse.size, response.errors.size) + + expectedResponse.foreach { case (topic, expectedError) => + assertEquals("The response error should match", expectedResponse(topic), errors(topic)) + // If no error validate topic exists + if (expectedError == Errors.NONE) { + validateTopicExists(topic) + } + } + } + + @Test + def testNotController() { + val request = new CreateTopicsRequest(Map("topic1" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, 1000) + val response = sendCreateTopicRequest(request, 0, notControllerSocketServer) + + val error = response.errors.asScala.head._2 + assertEquals("Expected controller error when routed incorrectly", Errors.NOT_CONTROLLER, error) + } + + private def validateTopicExists(topic: String): Unit = { + TestUtils.waitUntilMetadataIsPropagated(servers, topic, 0) + val metadata = sendMetadataRequest(new MetadataRequest(List(topic).asJava)).topicMetadata.asScala + assertTrue("The topic should be created", metadata.exists(p => p.topic.equals(topic) && p.error() == Errors.NONE)) + } + + private def replicaAssignmentToJava(assignments: Map[Int, List[Int]]) = { + assignments.map { case (k, v) => (k:Integer, v.map { i => i:Integer }.asJava) }.asJava + } + + private def sendCreateTopicRequest(request: CreateTopicsRequest, version: Short, socketServer: SocketServer = controllerSocketServer): CreateTopicsResponse = { + val response = send(request, ApiKeys.CREATE_TOPICS, version, socketServer) + CreateTopicsResponse.parse(response, version) + } + + private def sendMetadataRequest(request: MetadataRequest): MetadataResponse = { + val version = ProtoUtils.latestVersion(ApiKeys.METADATA.id) + val response = send(request, ApiKeys.METADATA, version) + MetadataResponse.parse(response, version) + } +} diff --git a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala index f80fa7d1a9265..07cedc0257043 100644 --- a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala @@ -89,10 +89,7 @@ class ProduceRequestTest extends BaseRequestTest { } private def sendProduceRequest(leaderId: Int, request: ProduceRequest): ProduceResponse = { - val socket = connect(s = servers.find(_.config.brokerId == leaderId).map(_.socketServer).getOrElse { - fail(s"Could not find broker with id $leaderId") - }) - val response = send(socket, request, ApiKeys.PRODUCE, ProtoUtils.latestVersion(ApiKeys.PRODUCE.id)) + val response = send(request, ApiKeys.PRODUCE, ProtoUtils.latestVersion(ApiKeys.PRODUCE.id), brokerSocketServer(leaderId)) ProduceResponse.parse(response) } diff --git a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala index 85570087a3d48..4e4fb95af5344 100644 --- a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala @@ -78,12 +78,12 @@ class SaslApiVersionsRequestTest extends BaseRequestTest with SaslTestHarness { } private def sendApiVersionsRequest(socket: Socket, request: ApiVersionsRequest, version: Short): ApiVersionsResponse = { - val response = send(socket, request, ApiKeys.API_VERSIONS, version) + val response = send(request, ApiKeys.API_VERSIONS, version, socket) ApiVersionsResponse.parse(response) } private def sendSaslHandshakeRequestValidateResponse(socket: Socket) { - val response = send(socket, new SaslHandshakeRequest("PLAIN"), ApiKeys.SASL_HANDSHAKE, 0) + val response = send(new SaslHandshakeRequest("PLAIN"), ApiKeys.SASL_HANDSHAKE, 0.toShort, socket) val handshakeResponse = SaslHandshakeResponse.parse(response) assertEquals(Errors.NONE.code, handshakeResponse.errorCode()) assertEquals(Collections.singletonList("PLAIN"), handshakeResponse.enabledMechanisms()) From 390f215608ebe8e9152f94cc9938f2d00d705b7f Mon Sep 17 00:00:00 2001 From: Grant Henke Date: Thu, 7 Jul 2016 15:42:19 -0500 Subject: [PATCH 2/5] Address Jun's Reviews --- .../kafka/common/protocol/Protocol.java | 2 +- .../common/requests/CreateTopicsRequest.java | 15 +++--- .../common/requests/CreateTopicsResponse.java | 13 +++--- .../scala/kafka/common/ErrorMapping.scala | 8 ++++ .../kafka/controller/KafkaController.scala | 12 ----- .../scala/kafka/server/AdminManager.scala | 19 ++++++-- .../kafka/server/DelayedCreateTopics.scala | 5 +- .../main/scala/kafka/server/KafkaApis.scala | 7 ++- .../main/scala/kafka/server/KafkaServer.scala | 2 +- .../kafka/server/ApiVersionsRequestTest.scala | 4 +- .../unit/kafka/server/BaseRequestTest.scala | 18 ++++++-- .../server/CreateTopicsRequestTest.scala | 46 +++++++++++-------- .../kafka/server/MetadataRequestTest.scala | 2 +- .../kafka/server/ProduceRequestTest.scala | 2 +- 14 files changed, 90 insertions(+), 65 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java index e2064e12e67aa..2610e048538c4 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java @@ -804,7 +804,7 @@ public class Protocol { "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 immediatly")); + "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", diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java index d5ce56e6988b7..3977835571174 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java @@ -24,6 +24,7 @@ import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -66,21 +67,21 @@ private TopicDetails(int numPartitions, public TopicDetails(int partitions, short replicationFactor, Map configs) { - this(partitions, replicationFactor, new HashMap>(), configs); + this(partitions, replicationFactor, Collections.>emptyMap(), configs); } public TopicDetails(int partitions, short replicationFactor) { - this(partitions, replicationFactor, new HashMap()); + this(partitions, replicationFactor, Collections.emptyMap()); } public TopicDetails(Map> replicasAssignments, Map configs) { - this(NO_NUM_PARTITIONS_SIGN, NO_REPLICATION_FACTOR_SIGN, replicasAssignments, configs); + this(NO_NUM_PARTITIONS, NO_REPLICATION_FACTOR, replicasAssignments, configs); } public TopicDetails(Map> replicasAssignments) { - this(replicasAssignments, new HashMap()); + this(replicasAssignments, Collections.emptyMap()); } } @@ -91,8 +92,8 @@ public TopicDetails(Map> replicasAssignments) { // This allows the broker to return an error code for these topics. private final Set duplicateTopics; - public static final int NO_NUM_PARTITIONS_SIGN = -1; - public static final short NO_REPLICATION_FACTOR_SIGN = -1; + public static final int NO_NUM_PARTITIONS = -1; + public static final short NO_REPLICATION_FACTOR = -1; public CreateTopicsRequest(Map topics, Integer timeout) { super(new Struct(CURRENT_SCHEMA)); @@ -134,7 +135,7 @@ public CreateTopicsRequest(Map topics, Integer timeout) { this.topics = topics; this.timeout = timeout; - this.duplicateTopics = new HashSet<>(); + this.duplicateTopics = Collections.emptySet(); } public CreateTopicsRequest(Struct struct) { diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java index a4fe431a41817..1e6d11e39a501 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java @@ -41,12 +41,13 @@ public class CreateTopicsResponse extends AbstractRequestResponse { * * INVALID_TOPIC_EXCEPTION(17) * CLUSTER_AUTHORIZATION_FAILED(31) - * TOPIC_ALREADY_EXISTS(32) - * INVALID_PARTITIONS(38) - * INVALID_REPLICATION_FACTOR(39) - * INVALID_REPLICA_ASSIGNMENT(40) - * INVALID_CONFIG(41) - * NOT_CONTROLLER(42) + * TOPIC_ALREADY_EXISTS(36) + * INVALID_PARTITIONS(37) + * INVALID_REPLICATION_FACTOR(38) + * INVALID_REPLICA_ASSIGNMENT(39) + * INVALID_CONFIG(40) + * NOT_CONTROLLER(41) + * INVALID_REQUEST(42) */ private final Map errors; diff --git a/core/src/main/scala/kafka/common/ErrorMapping.scala b/core/src/main/scala/kafka/common/ErrorMapping.scala index bbbc854c8d4fc..be5fed2049d2f 100644 --- a/core/src/main/scala/kafka/common/ErrorMapping.scala +++ b/core/src/main/scala/kafka/common/ErrorMapping.scala @@ -66,6 +66,14 @@ object ErrorMapping { // 32: INVALID_TIMESTAMP // 33: UNSUPPORTED_SASL_MECHANISM // 34: ILLEGAL_SASL_STATE + // 35: UNSUPPORTED_VERSION + // 36: TOPIC_ALREADY_EXISTS + // 37: INVALID_PARTITIONS + // 38: INVALID_REPLICATION_FACTOR + // 39: INVALID_REPLICA_ASSIGNMENT + // 40: INVALID_CONFIG + // 41: NOT_CONTROLLER + // 42: INVALID_REQUEST private val exceptionToCode = Map[Class[Throwable], Short]( diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index 7548cc1957314..1584cc95a38ee 100755 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/core/src/main/scala/kafka/controller/KafkaController.scala @@ -180,8 +180,6 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat private val preferredReplicaElectionListener = new PreferredReplicaElectionListener(this) private val isrChangeNotificationListener = new IsrChangeNotificationListener(this) - val topicPurgatory = DelayedOperationPurgatory[DelayedOperation]("topic", config.brokerId) - newGauge( "ActiveControllerCount", new Gauge[Int] { @@ -227,15 +225,6 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat "id_%d-host_%s-port_%d".format(config.brokerId, controllerListener.get.host, controllerListener.get.port) } - /** - * Try to complete delayed topic operations with the request key - */ - def tryCompleteDelayedTopicOperations(topic: String) { - val key = TopicKey(topic) - val completed = topicPurgatory.checkAndComplete(key) - debug(s"Request key ${key.keyLabel} unblocked $completed topic requests.") - } - /** * On clean shutdown, the controller first determines the partitions that the * shutting down broker leads, and moves leadership of those partitions to another broker @@ -517,7 +506,6 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat // subscribe to partition changes topics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic)) onNewPartitionCreation(newPartitions) - topics.foreach(topic => tryCompleteDelayedTopicOperations(topic)) } /** diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala index bee8c48df44d1..29551710575e2 100644 --- a/core/src/main/scala/kafka/server/AdminManager.scala +++ b/core/src/main/scala/kafka/server/AdminManager.scala @@ -19,7 +19,6 @@ package kafka.server import java.util.Properties import kafka.admin.AdminUtils -import kafka.controller.KafkaController import kafka.log.LogConfig import kafka.metrics.KafkaMetricsGroup import kafka.utils._ @@ -33,11 +32,21 @@ import scala.collection.JavaConverters._ class AdminManager(val config: KafkaConfig, val metrics: Metrics, - val controller: KafkaController, val metadataCache: MetadataCache, val zkUtils: ZkUtils) extends Logging with KafkaMetricsGroup { this.logIdent = "[Admin Manager on Broker " + config.brokerId + "]: " + val topicPurgatory = DelayedOperationPurgatory[DelayedOperation]("topic", config.brokerId) + + /** + * Try to complete delayed topic operations with the request key + */ + def tryCompleteDelayedTopicOperations(topic: String) { + val key = TopicKey(topic) + val completed = topicPurgatory.checkAndComplete(key) + debug(s"Request key ${key.keyLabel} unblocked $completed topic requests.") + } + /** * Create topics and wait until the topics have been completely created. * The callback function will be triggered either when timeout, error or the topics are created. @@ -57,7 +66,7 @@ class AdminManager(val config: KafkaConfig, LogConfig.validate(configs) val assignments = { - if ((arguments.numPartitions != NO_NUM_PARTITIONS_SIGN || arguments.replicationFactor != NO_REPLICATION_FACTOR_SIGN) + if ((arguments.numPartitions != NO_NUM_PARTITIONS || arguments.replicationFactor != NO_REPLICATION_FACTOR) && !arguments.replicasAssignments.isEmpty) throw new InvalidRequestException("Both numPartitions or replicationFactor and replicasAssignments were set. " + "Both cannot be used at the same time.") @@ -76,7 +85,7 @@ class AdminManager(val config: KafkaConfig, CreateTopicMetadata(topic, assignments, Errors.NONE) } catch { case e: Throwable => - error(s"Error processing create topic request for topic $topic with arguments $arguments", e) + warn(s"Error processing create topic request for topic $topic with arguments $arguments", e) CreateTopicMetadata(topic, Map(), Errors.forException(e)) } } @@ -97,7 +106,7 @@ class AdminManager(val config: KafkaConfig, val delayedCreate = new DelayedCreateTopics(timeout, metadata.toSeq, this, responseCallback) val delayedCreateKeys = createInfo.keys.map(new TopicKey(_)).toSeq // try to complete the request immediately, otherwise put it into the purgatory - controller.topicPurgatory.tryCompleteElseWatch(delayedCreate, delayedCreateKeys) + topicPurgatory.tryCompleteElseWatch(delayedCreate, delayedCreateKeys) } } } diff --git a/core/src/main/scala/kafka/server/DelayedCreateTopics.scala b/core/src/main/scala/kafka/server/DelayedCreateTopics.scala index 720e109cc68e1..b67746f716d1f 100644 --- a/core/src/main/scala/kafka/server/DelayedCreateTopics.scala +++ b/core/src/main/scala/kafka/server/DelayedCreateTopics.scala @@ -47,7 +47,6 @@ class DelayedCreateTopics(delayMs: Long, */ override def tryComplete() : Boolean = { trace(s"Trying to complete operation for $createMetadata") - trace(s"Live brokers ${adminManager.controller.controllerContext.liveBrokerIds}") val leaderlessPartitionCount = createMetadata.foldLeft(0) { case (topicCounter, metadata) => topicCounter + missingLeaderCount(metadata.topic, metadata.replicaAssignments.keySet) @@ -86,9 +85,7 @@ class DelayedCreateTopics(delayMs: Long, } private def isMissingLeader(topic: String, partition: Int): Boolean = { - val leadershipInfo = adminManager.controller.controllerContext.partitionLeadershipInfo - val leaderInfo =leadershipInfo.get(TopicAndPartition(topic, partition)) val partitionInfo = adminManager.metadataCache.getPartitionInfo(topic, partition) - leaderInfo.isEmpty || leaderInfo.get.leaderAndIsr.leader == LeaderAndIsr.NoLeader + partitionInfo.isEmpty || partitionInfo.get.leaderIsrAndControllerEpoch.leaderAndIsr.leader == LeaderAndIsr.NoLeader } } diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 72bc7cd54460b..d8b22521cfe73 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -185,6 +185,11 @@ class KafkaApis(val requestChannel: RequestChannel, val updateMetadataResponse = if (authorize(request.session, ClusterAction, Resource.ClusterResource)) { replicaManager.maybeUpdateMetadataCache(correlationId, updateMetadataRequest, metadataCache) + if(controller.isActive && !updateMetadataRequest.partitionStates.isEmpty) { + updateMetadataRequest.partitionStates.asScala.keys.map(_.topic).toSet { topic => + adminManager.tryCompleteDelayedTopicOperations(topic) + } + } new UpdateMetadataResponse(Errors.NONE.code) } else { new UpdateMetadataResponse(Errors.CLUSTER_AUTHORIZATION_FAILED.code) @@ -1055,7 +1060,7 @@ class KafkaApis(val requestChannel: RequestChannel, def sendResponseCallback(results: Map[String, Errors]): Unit = { val respHeader = new ResponseHeader(request.header.correlationId) if(duplicateTopics.nonEmpty) - error("Multiple entries for the same topic were found for the following topics: " + duplicateTopics.keySet.mkString(",")) + warn(s"Create topics request from client ${request.header.clientId} contains multiple entries for the following topics: ${duplicateTopics.keySet.mkString(",")}") val completeResults = results ++ duplicateTopics.keySet.map((_, Errors.INVALID_REQUEST)).toMap val responseBody = new CreateTopicsResponse(completeResults.asJava) trace(s"Sending create topics response $responseBody for correlation id ${request.header.correlationId} to client ${request.header.clientId}.") diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 07f8dd1a02512..715a1dc928013 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -201,7 +201,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr kafkaController.startup() /* start admin manager */ - adminManager = new AdminManager(config, metrics, kafkaController, metadataCache, zkUtils) + adminManager = new AdminManager(config, metrics, metadataCache, zkUtils) /* start group coordinator */ groupCoordinator = GroupCoordinator(config, zkUtils, replicaManager, kafkaMetricsTime) diff --git a/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala index f2dd60f24690a..fdaf2f59cb04a 100644 --- a/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala @@ -55,7 +55,7 @@ class ApiVersionsRequestTest extends BaseRequestTest { } private def sendApiVersionsRequest(request: ApiVersionsRequest, version: Short): ApiVersionsResponse = { - val response = send(request, ApiKeys.API_VERSIONS, version) + val response = send(request, ApiKeys.API_VERSIONS, Some(version)) ApiVersionsResponse.parse(response) } -} \ No newline at end of file +} diff --git a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala index 78c8f6e57f7cd..6909ed34c556f 100644 --- a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala @@ -25,12 +25,10 @@ import java.util.Properties import kafka.integration.KafkaServerTestHarness import kafka.network.SocketServer import kafka.utils._ -import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol} +import org.apache.kafka.common.protocol.{ApiKeys, ProtoUtils, SecurityProtocol} import org.apache.kafka.common.requests.{AbstractRequest, RequestHeader, ResponseHeader} import org.junit.Before -import scala.collection.mutable.Buffer - abstract class BaseRequestTest extends KafkaServerTestHarness { private var correlationId = 0 @@ -103,11 +101,21 @@ abstract class BaseRequestTest extends KafkaServerTestHarness { receiveResponse(socket) } - def send(request: AbstractRequest, apiKey: ApiKeys, version: Short, + /** + * + * @param request + * @param apiKey + * @param version An optional version to use when sending the request. If not set, the latest known version is used + * @param destination An optional SocketServer ot send the request to. If not set, any available server is used. + * @param protocol An optional SecurityProtocol to use. If not set, PLAINTEXT is used. + * @return + */ + def send(request: AbstractRequest, apiKey: ApiKeys, version: Option[Short] = None, destination: SocketServer = anySocketServer, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): ByteBuffer = { + val requestVersion = version.getOrElse(ProtoUtils.latestVersion(apiKey.id)) val socket = connect(destination, protocol) try { - send(request, apiKey, version, socket) + send(request, apiKey, requestVersion, socket) } finally { socket.close() } diff --git a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala index 6f815412411d9..3526933ae8f6a 100644 --- a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala @@ -60,23 +60,31 @@ class CreateTopicsRequestTest extends BaseRequestTest { assertTrue(s"There should be no errors, found ${response.errors.asScala}", error.isEmpty) request.topics.asScala.foreach { case (topic, details) => + + def verifyMetadata(socketServer: SocketServer) = { + val metadata = sendMetadataRequest(new MetadataRequest(List(topic).asJava)).topicMetadata.asScala + val metadataForTopic = metadata.filter(p => p.topic.equals(topic)).head + + val partitions = if (!details.replicasAssignments.isEmpty) + details.replicasAssignments.size + else + details.numPartitions + + val replication = if (!details.replicasAssignments.isEmpty) + details.replicasAssignments.asScala.head._2.size + else + details.replicationFactor + + assertNotNull("The topic should be created", metadataForTopic) + assertEquals("The topic should have the correct number of partitions", partitions, metadataForTopic.partitionMetadata.size) + assertEquals("The topic should have the correct replication factor", replication, metadataForTopic.partitionMetadata.asScala.head.replicas.size) + } + + // Verify controller broker has the correct metadata + verifyMetadata(controllerSocketServer) + // Wait until metadata is propagated and validate non-controller broker has the correct metadata TestUtils.waitUntilMetadataIsPropagated(servers, topic, 0) - val metadata = sendMetadataRequest(new MetadataRequest(List(topic).asJava)).topicMetadata.asScala - val metadataForTopic = metadata.filter(p => p.topic.equals(topic)).head - - val partitions = if (!details.replicasAssignments.isEmpty) - details.replicasAssignments.size - else - details.numPartitions - - val replication = if (!details.replicasAssignments.isEmpty) - details.replicasAssignments.asScala.head._2.size - else - details.replicationFactor - - assertNotNull("The topic should be created", metadataForTopic) - assertEquals("The topic should have the correct number of partitions", partitions, metadataForTopic.partitionMetadata.size) - assertEquals("The topic should have the correct replication factor", replication, metadataForTopic.partitionMetadata.asScala.head.replicas.size) + verifyMetadata(notControllerSocketServer) } } @@ -200,13 +208,13 @@ class CreateTopicsRequestTest extends BaseRequestTest { } private def sendCreateTopicRequest(request: CreateTopicsRequest, version: Short, socketServer: SocketServer = controllerSocketServer): CreateTopicsResponse = { - val response = send(request, ApiKeys.CREATE_TOPICS, version, socketServer) + val response = send(request, ApiKeys.CREATE_TOPICS, Some(version), socketServer) CreateTopicsResponse.parse(response, version) } - private def sendMetadataRequest(request: MetadataRequest): MetadataResponse = { + private def sendMetadataRequest(request: MetadataRequest, destination: SocketServer = anySocketServer): MetadataResponse = { val version = ProtoUtils.latestVersion(ApiKeys.METADATA.id) - val response = send(request, ApiKeys.METADATA, version) + val response = send(request, ApiKeys.METADATA, destination = destination) MetadataResponse.parse(response, version) } } diff --git a/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala b/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala index 3d4b40c6a9251..a5909fe74707d 100644 --- a/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala @@ -162,7 +162,7 @@ class MetadataRequestTest extends BaseRequestTest { } private def sendMetadataRequest(request: MetadataRequest, version: Short): MetadataResponse = { - val response = send(request, ApiKeys.METADATA, version) + val response = send(request, ApiKeys.METADATA, Some(version)) MetadataResponse.parse(response, version) } } diff --git a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala index 07cedc0257043..9e4c800e3063a 100644 --- a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala @@ -89,7 +89,7 @@ class ProduceRequestTest extends BaseRequestTest { } private def sendProduceRequest(leaderId: Int, request: ProduceRequest): ProduceResponse = { - val response = send(request, ApiKeys.PRODUCE, ProtoUtils.latestVersion(ApiKeys.PRODUCE.id), brokerSocketServer(leaderId)) + val response = send(request, ApiKeys.PRODUCE, destination = brokerSocketServer(leaderId)) ProduceResponse.parse(response) } From 5ce746f723fa25a56c838d239e19f952604a8fc6 Mon Sep 17 00:00:00 2001 From: Grant Henke Date: Mon, 11 Jul 2016 15:36:28 -0500 Subject: [PATCH 3/5] Address reviews --- .../kafka/server/DelayedCreateTopics.scala | 10 +++---- .../main/scala/kafka/server/KafkaApis.scala | 29 +++++++++++-------- .../main/scala/kafka/server/KafkaServer.scala | 1 - .../server/CreateTopicsRequestTest.scala | 12 ++++++++ 4 files changed, 34 insertions(+), 18 deletions(-) diff --git a/core/src/main/scala/kafka/server/DelayedCreateTopics.scala b/core/src/main/scala/kafka/server/DelayedCreateTopics.scala index b67746f716d1f..b74b596a4567f 100644 --- a/core/src/main/scala/kafka/server/DelayedCreateTopics.scala +++ b/core/src/main/scala/kafka/server/DelayedCreateTopics.scala @@ -18,7 +18,6 @@ package kafka.server import kafka.api.LeaderAndIsr -import kafka.common.TopicAndPartition import org.apache.kafka.common.protocol.Errors import scala.collection._ @@ -42,15 +41,16 @@ class DelayedCreateTopics(delayMs: Long, extends DelayedOperation(delayMs) { /** - * The operation can be completed if all of the topics exist every partition has a leader in the controller. + * The operation can be completed if all of the topics that do not have an error exist and every partition has a leader in the controller. * See KafkaController.onNewTopicCreation */ override def tryComplete() : Boolean = { trace(s"Trying to complete operation for $createMetadata") - val leaderlessPartitionCount = createMetadata.foldLeft(0) { case (topicCounter, metadata) => - topicCounter + missingLeaderCount(metadata.topic, metadata.replicaAssignments.keySet) - } + val leaderlessPartitionCount = createMetadata.filter(_.error == Errors.NONE) + .foldLeft(0) { case (topicCounter, metadata) => + topicCounter + missingLeaderCount(metadata.topic, metadata.replicaAssignments.keySet) + } if (leaderlessPartitionCount == 0) { trace("All partitions have a leader, completing the delayed operation") diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index d8b22521cfe73..e7c56244d39e6 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -185,8 +185,8 @@ class KafkaApis(val requestChannel: RequestChannel, val updateMetadataResponse = if (authorize(request.session, ClusterAction, Resource.ClusterResource)) { replicaManager.maybeUpdateMetadataCache(correlationId, updateMetadataRequest, metadataCache) - if(controller.isActive && !updateMetadataRequest.partitionStates.isEmpty) { - updateMetadataRequest.partitionStates.asScala.keys.map(_.topic).toSet { topic => + if (controller.isActive && !updateMetadataRequest.partitionStates.isEmpty) { + updateMetadataRequest.partitionStates.keySet.asScala.map(_.topic).foreach { topic => adminManager.tryCompleteDelayedTopicOperations(topic) } } @@ -1053,21 +1053,14 @@ class KafkaApis(val requestChannel: RequestChannel, def handleCreateTopicsRequest(request: RequestChannel.Request) { val createTopicsRequest = request.body.asInstanceOf[CreateTopicsRequest] - val (validTopics, duplicateTopics) = createTopicsRequest.topics.asScala.partition { case (topic, _) => - !createTopicsRequest.duplicateTopics.contains(topic) - } - def sendResponseCallback(results: Map[String, Errors]): Unit = { val respHeader = new ResponseHeader(request.header.correlationId) - if(duplicateTopics.nonEmpty) - warn(s"Create topics request from client ${request.header.clientId} contains multiple entries for the following topics: ${duplicateTopics.keySet.mkString(",")}") - val completeResults = results ++ duplicateTopics.keySet.map((_, Errors.INVALID_REQUEST)).toMap - val responseBody = new CreateTopicsResponse(completeResults.asJava) + val responseBody = new CreateTopicsResponse(results.asJava) trace(s"Sending create topics response $responseBody for correlation id ${request.header.correlationId} to client ${request.header.clientId}.") requestChannel.sendResponse(new RequestChannel.Response(request, new ResponseSend(request.connectionId, respHeader, responseBody))) } - if(!controller.isActive()) { + if (!controller.isActive()) { val results = createTopicsRequest.topics.asScala.map { case (topic, _) => (topic, Errors.NOT_CONTROLLER) } @@ -1079,10 +1072,22 @@ class KafkaApis(val requestChannel: RequestChannel, sendResponseCallback(results) } else { + val (validTopics, duplicateTopics) = createTopicsRequest.topics.asScala.partition { case (topic, _) => + !createTopicsRequest.duplicateTopics.contains(topic) + } + + // Special handling to add duplicate topics to the response + def sendResponseWithDuplicatesCallback(results: Map[String, Errors]): Unit = { + if (duplicateTopics.nonEmpty) + warn(s"Create topics request from client ${request.header.clientId} contains multiple entries for the following topics: ${duplicateTopics.keySet.mkString(",")}") + val completeResults = results ++ duplicateTopics.keySet.map((_, Errors.INVALID_REQUEST)).toMap + sendResponseCallback(completeResults) + } + adminManager.createTopics( createTopicsRequest.timeout.toInt, validTopics, - sendResponseCallback + sendResponseWithDuplicatesCallback ) } } diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 715a1dc928013..78c660649e0a4 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -200,7 +200,6 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr kafkaController = new KafkaController(config, zkUtils, brokerState, kafkaMetricsTime, metrics, threadNamePrefix) kafkaController.startup() - /* start admin manager */ adminManager = new AdminManager(config, metrics, metadataCache, zkUtils) /* start group coordinator */ diff --git a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala index 3526933ae8f6a..46e39b8c11318 100644 --- a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala @@ -114,11 +114,13 @@ class CreateTopicsRequestTest extends BaseRequestTest { existingTopic -> new CreateTopicsRequest.TopicDetails(1, 1.toShort), "partial-partitions" -> new CreateTopicsRequest.TopicDetails(-1, 1.toShort), "partial-replication" -> new CreateTopicsRequest.TopicDetails(1, (numBrokers + 1).toShort), + "partial-assignment" -> new CreateTopicsRequest.TopicDetails(invalidAssignments), "partial-none" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, timeout), Map( existingTopic -> Errors.TOPIC_ALREADY_EXISTS, "partial-partitions" -> Errors.INVALID_PARTITIONS, "partial-replication" -> Errors.INVALID_REPLICATION_FACTOR, + "partial-assignment" -> Errors.INVALID_REPLICA_ASSIGNMENT, "partial-none" -> Errors.NONE ) ) @@ -149,6 +151,16 @@ class CreateTopicsRequestTest extends BaseRequestTest { assertFalse("Request doesn't have duplicate topics", duplicateRequest.duplicateTopics().isEmpty) validateErrorCreateTopicsRequests(duplicateRequest, Map("duplicate-topic" -> Errors.INVALID_REQUEST)) + // Duplicate Partial + val doubleRequest = new CreateTopicsRequest(Map( + "duplicate-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort), + "other-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, 1000) + val duplicateDoubleRequest = duplicateFirstTopic(doubleRequest) + assertFalse("Request doesn't have duplicate topics", duplicateDoubleRequest.duplicateTopics().isEmpty) + validateErrorCreateTopicsRequests(duplicateDoubleRequest, Map( + "duplicate-topic" -> Errors.INVALID_REQUEST, + "other-topic" -> Errors.NONE)) + // Partitions/ReplicationFactor and ReplicaAssignment val assignments = replicaAssignmentToJava(Map(0 -> List(0))) val assignmentRequest = new CreateTopicsRequest(Map("bad-args-topic" -> new CreateTopicsRequest.TopicDetails(assignments)).asJava, 1000) From 21bf32c85a253e87e4c97071971e0ff27d45e622 Mon Sep 17 00:00:00 2001 From: Grant Henke Date: Tue, 12 Jul 2016 08:29:07 -0500 Subject: [PATCH 4/5] Remove controller.isActive check --- core/src/main/scala/kafka/server/AdminManager.scala | 2 ++ core/src/main/scala/kafka/server/KafkaApis.scala | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala index 29551710575e2..fc3a7f03780c6 100644 --- a/core/src/main/scala/kafka/server/AdminManager.scala +++ b/core/src/main/scala/kafka/server/AdminManager.scala @@ -38,6 +38,8 @@ class AdminManager(val config: KafkaConfig, val topicPurgatory = DelayedOperationPurgatory[DelayedOperation]("topic", config.brokerId) + def hasDelayedTopicOperations = topicPurgatory.delayed() != 0 + /** * Try to complete delayed topic operations with the request key */ diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index e7c56244d39e6..5cadb8bee062c 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -185,7 +185,7 @@ class KafkaApis(val requestChannel: RequestChannel, val updateMetadataResponse = if (authorize(request.session, ClusterAction, Resource.ClusterResource)) { replicaManager.maybeUpdateMetadataCache(correlationId, updateMetadataRequest, metadataCache) - if (controller.isActive && !updateMetadataRequest.partitionStates.isEmpty) { + if (adminManager.hasDelayedTopicOperations) { updateMetadataRequest.partitionStates.keySet.asScala.map(_.topic).foreach { topic => adminManager.tryCompleteDelayedTopicOperations(topic) } From 8bedd81f8c8770f8651a448e21bcf7a5d07bd2d2 Mon Sep 17 00:00:00 2001 From: Grant Henke Date: Tue, 12 Jul 2016 08:38:59 -0500 Subject: [PATCH 5/5] Add timeout test comment --- .../test/scala/unit/kafka/server/CreateTopicsRequestTest.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala index 46e39b8c11318..52809b3b99e00 100644 --- a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala @@ -127,6 +127,7 @@ class CreateTopicsRequestTest extends BaseRequestTest { validateTopicExists("partial-none") // Timeout + // We don't expect a request to ever complete within 1ms. A timeout of 1 ms allows us to test the purgatory timeout logic. validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-timeout" -> new CreateTopicsRequest.TopicDetails(10, 3.toShort)).asJava, 1), Map("error-timeout" -> Errors.REQUEST_TIMED_OUT)) validateErrorCreateTopicsRequests(new CreateTopicsRequest(Map("error-timeout-zero" -> new CreateTopicsRequest.TopicDetails(10, 3.toShort)).asJava, 0),