diff --git a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java index df0e6d5105ca9..f619fc99a0b30 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java +++ b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java @@ -59,4 +59,14 @@ public long writeTo(GatheringByteChannel channel) throws IOException { remaining -= written; return written; } + + // Used only by BlockingChannel, so we may be able to get rid of this when/if we get rid of BlockingChannel + public long writeCompletelyTo(GatheringByteChannel channel) throws IOException { + long totalWritten = 0L; + while (!completed()) { + long written = writeTo(channel); + totalWritten += written; + } + return totalWritten; + } } 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 3dc8b015afd23..5aaf9e0a6d3d9 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 @@ -16,16 +16,12 @@ */ package org.apache.kafka.common.protocol; -import static org.apache.kafka.common.protocol.types.Type.BYTES; -import static org.apache.kafka.common.protocol.types.Type.INT16; -import static org.apache.kafka.common.protocol.types.Type.INT32; -import static org.apache.kafka.common.protocol.types.Type.INT64; -import static org.apache.kafka.common.protocol.types.Type.STRING; - import org.apache.kafka.common.protocol.types.ArrayOf; import org.apache.kafka.common.protocol.types.Field; import org.apache.kafka.common.protocol.types.Schema; +import static org.apache.kafka.common.protocol.types.Type.*; + public class Protocol { public static final Schema REQUEST_HEADER = new Schema(new Field("api_key", INT16, "The id of the request type."), @@ -420,6 +416,27 @@ public class Protocol { public static final Schema[] HEARTBEAT_REQUEST = new Schema[] {HEARTBEAT_REQUEST_V0}; public static final Schema[] HEARTBEAT_RESPONSE = new Schema[] {HEARTBEAT_RESPONSE_V0}; + /* Replica api */ + public static final Schema STOP_REPLICA_REQUEST_PARTITION_V0 = new Schema(new Field("topic", STRING, "The partition name."), + new Field("partition", INT32, "Topic partition id.")); + + public static final Schema STOP_REPLICA_REQUEST_V0 = new Schema(new Field("controller_id", INT32, "The controller id."), + new Field("controller_epoch", INT32, "The controller epoch."), + new Field("delete_partitions", INT8), + new Field("partitions", + new ArrayOf(STOP_REPLICA_REQUEST_PARTITION_V0))); + + public static final Schema STOP_REPLICA_RESPONSE_PARTITION_V0 = new Schema(new Field("topic", STRING, "The partition name."), + new Field("partition", INT32, "Topic partition id."), + new Field("error_code", INT16)); + + public static final Schema STOP_REPLICA_RESPONSE_V0 = new Schema(new Field("error_code", INT16), + new Field("partitions", + new ArrayOf(STOP_REPLICA_RESPONSE_PARTITION_V0))); + + public static final Schema[] STOP_REPLICA_REQUEST = new Schema[] {STOP_REPLICA_REQUEST_V0}; + public static final Schema[] STOP_REPLICA_RESPONSE = new Schema[] {STOP_REPLICA_RESPONSE_V0}; + /* an array of all requests and responses with all schema versions */ public static final Schema[][] REQUESTS = new Schema[ApiKeys.MAX_API_KEY + 1][]; public static final Schema[][] RESPONSES = new Schema[ApiKeys.MAX_API_KEY + 1][]; @@ -433,7 +450,7 @@ public class Protocol { REQUESTS[ApiKeys.LIST_OFFSETS.id] = LIST_OFFSET_REQUEST; REQUESTS[ApiKeys.METADATA.id] = METADATA_REQUEST; REQUESTS[ApiKeys.LEADER_AND_ISR.id] = new Schema[] {}; - REQUESTS[ApiKeys.STOP_REPLICA.id] = new Schema[] {}; + REQUESTS[ApiKeys.STOP_REPLICA.id] = STOP_REPLICA_REQUEST; REQUESTS[ApiKeys.UPDATE_METADATA_KEY.id] = new Schema[] {}; REQUESTS[ApiKeys.CONTROLLED_SHUTDOWN_KEY.id] = new Schema[] {}; REQUESTS[ApiKeys.OFFSET_COMMIT.id] = OFFSET_COMMIT_REQUEST; @@ -442,12 +459,13 @@ public class Protocol { REQUESTS[ApiKeys.JOIN_GROUP.id] = JOIN_GROUP_REQUEST; REQUESTS[ApiKeys.HEARTBEAT.id] = HEARTBEAT_REQUEST; + RESPONSES[ApiKeys.PRODUCE.id] = PRODUCE_RESPONSE; RESPONSES[ApiKeys.FETCH.id] = FETCH_RESPONSE; RESPONSES[ApiKeys.LIST_OFFSETS.id] = LIST_OFFSET_RESPONSE; RESPONSES[ApiKeys.METADATA.id] = METADATA_RESPONSE; RESPONSES[ApiKeys.LEADER_AND_ISR.id] = new Schema[] {}; - RESPONSES[ApiKeys.STOP_REPLICA.id] = new Schema[] {}; + RESPONSES[ApiKeys.STOP_REPLICA.id] = STOP_REPLICA_RESPONSE; RESPONSES[ApiKeys.UPDATE_METADATA_KEY.id] = new Schema[] {}; RESPONSES[ApiKeys.CONTROLLED_SHUTDOWN_KEY.id] = new Schema[] {}; RESPONSES[ApiKeys.OFFSET_COMMIT.id] = OFFSET_COMMIT_RESPONSE; 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 5d3d52859587c..e3169572b4a89 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 @@ -55,8 +55,10 @@ public static AbstractRequest getRequest(int requestId, int versionId, ByteBuffe return JoinGroupRequest.parse(buffer, versionId); case HEARTBEAT: return HeartbeatRequest.parse(buffer, versionId); + case STOP_REPLICA: + return StopReplicaRequest.parse(buffer, versionId); default: return null; } } -} \ No newline at end of file +} diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java new file mode 100644 index 0000000000000..b8be91d6b767c --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java @@ -0,0 +1,120 @@ +/** + * 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.TopicPartition; +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.*; + +public class StopReplicaRequest extends AbstractRequest { + private static final Schema CURRENT_SCHEMA = ProtoUtils.currentRequestSchema(ApiKeys.STOP_REPLICA.id); + + private static final String CONTROLLER_ID_KEY_NAME = "controller_id"; + private static final String CONTROLLER_EPOCH_KEY_NAME = "controller_epoch"; + private static final String DELETE_PARTITIONS_KEY_NAME = "delete_partitions"; + private static final String PARTITIONS_KEY_NAME = "partitions"; + + private static final String TOPIC_KEY_NAME = "topic"; + private static final String PARTITION_KEY_NAME = "partition"; + + private final int controllerId; + private final int controllerEpoch; + private final boolean deletePartitions; + private final Set partitions; + + public StopReplicaRequest(int controllerId, int controllerEpoch, boolean deletePartitions, Set partitions) { + super(new Struct(CURRENT_SCHEMA)); + + struct.set(CONTROLLER_ID_KEY_NAME, controllerId); + struct.set(CONTROLLER_EPOCH_KEY_NAME, controllerEpoch); + struct.set(DELETE_PARTITIONS_KEY_NAME, deletePartitions ? (byte) 1 : (byte) 0); + + List partitionDatas = new ArrayList<>(partitions.size()); + for (TopicPartition partition : partitions) { + Struct partitionData = struct.instance(PARTITIONS_KEY_NAME); + partitionData.set(TOPIC_KEY_NAME, partition.topic()); + partitionData.set(PARTITION_KEY_NAME, partition.partition()); + partitionDatas.add(partitionData); + } + + struct.set(PARTITIONS_KEY_NAME, partitionDatas.toArray()); + + this.controllerId = controllerId; + this.controllerEpoch = controllerEpoch; + this.deletePartitions = deletePartitions; + this.partitions = partitions; + } + + public StopReplicaRequest(Struct struct) { + super(struct); + + partitions = new HashSet<>(); + for (Object partitionDataObj : struct.getArray(PARTITIONS_KEY_NAME)) { + Struct partitionData = (Struct) partitionDataObj; + String topic = partitionData.getString(TOPIC_KEY_NAME); + int partition = partitionData.getInt(PARTITION_KEY_NAME); + partitions.add(new TopicPartition(topic, partition)); + } + + controllerId = struct.getInt(CONTROLLER_ID_KEY_NAME); + controllerEpoch = struct.getInt(CONTROLLER_EPOCH_KEY_NAME); + deletePartitions = ((byte) struct.get(DELETE_PARTITIONS_KEY_NAME)) != 0; + } + + @Override + public AbstractRequestResponse getErrorResponse(int versionId, Throwable e) { + Map responses = new HashMap<>(partitions.size()); + for (TopicPartition partition : partitions) { + responses.put(partition, Errors.forException(e).code()); + } + + switch (versionId) { + case 0: + return new StopReplicaResponse(responses, Errors.NONE.code()); + 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.STOP_REPLICA.id))); + } + } + + public int controllerId() { + return controllerId; + } + + public int controllerEpoch() { + return controllerEpoch; + } + + public boolean deletePartitions() { + return deletePartitions; + } + + public Set partitions() { + return partitions; + } + + public static StopReplicaRequest parse(ByteBuffer buffer, int versionId) { + return new StopReplicaRequest(ProtoUtils.parseRequest(ApiKeys.STOP_REPLICA.id, versionId, buffer)); + } + + public static StopReplicaRequest parse(ByteBuffer buffer) { + return new StopReplicaRequest((Struct) CURRENT_SCHEMA.read(buffer)); + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java new file mode 100644 index 0000000000000..6ab266bf22179 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.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.TopicPartition; +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.*; + +public class StopReplicaResponse extends AbstractRequestResponse { + private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.STOP_REPLICA.id); + + private static final String ERROR_CODE_KEY_NAME = "error_code"; + private static final String PARTITIONS_KEY_NAME = "partitions"; + + private static final String PARTITIONS_TOPIC_KEY_NAME = "topic"; + private static final String PARTITIONS_PARTITION_KEY_NAME = "partition"; + private static final String PARTITIONS_ERROR_CODE_KEY_NAME = "error_code"; + + private final Map responses; + private final short errorCode; + + public StopReplicaResponse(Map responses) { + this(responses, Errors.NONE.code()); + } + + public StopReplicaResponse(Map responses, short errorCode) { + super(new Struct(CURRENT_SCHEMA)); + + struct.set(ERROR_CODE_KEY_NAME, errorCode); + + List responseDatas = new ArrayList<>(responses.size()); + for (Map.Entry response : responses.entrySet()) { + Struct partitionData = struct.instance(PARTITIONS_KEY_NAME); + TopicPartition partition = response.getKey(); + partitionData.set(PARTITIONS_TOPIC_KEY_NAME, partition.topic()); + partitionData.set(PARTITIONS_PARTITION_KEY_NAME, partition.partition()); + partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue()); + responseDatas.add(partitionData); + } + + struct.set(PARTITIONS_KEY_NAME, responseDatas.toArray()); + struct.set(ERROR_CODE_KEY_NAME, errorCode); + + this.responses = responses; + this.errorCode = errorCode; + } + + public StopReplicaResponse(Struct struct) { + super(struct); + + responses = new HashMap<>(); + for (Object responseDataObj : struct.getArray(PARTITIONS_KEY_NAME)) { + Struct responseData = (Struct) responseDataObj; + String topic = responseData.getString(PARTITIONS_TOPIC_KEY_NAME); + int partition = responseData.getInt(PARTITIONS_PARTITION_KEY_NAME); + short errorCode = responseData.getShort(PARTITIONS_ERROR_CODE_KEY_NAME); + responses.put(new TopicPartition(topic, partition), errorCode); + } + + errorCode = struct.getShort(ERROR_CODE_KEY_NAME); + } + + public Map responses() { + return responses; + } + + public short errorCode() { + return errorCode; + } + + public static StopReplicaResponse parse(ByteBuffer buffer, int versionId) { + return new StopReplicaResponse(ProtoUtils.parseRequest(ApiKeys.STOP_REPLICA.id, versionId, buffer)); + } + + public static StopReplicaResponse parse(ByteBuffer buffer) { + return new StopReplicaResponse((Struct) CURRENT_SCHEMA.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 8b2aca85fa738..37fedb6a9e74b 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 @@ -23,10 +23,7 @@ import java.lang.reflect.Method; import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import static org.junit.Assert.assertEquals; @@ -63,7 +60,10 @@ public void testSerialization() throws Exception { createOffsetFetchResponse(), createProduceRequest(), createProduceRequest().getErrorResponse(0, new UnknownServerException()), - createProduceResponse()); + createProduceResponse(), + createStopReplicaRequest(), + createStopReplicaRequest().getErrorResponse(0, new UnknownServerException()), + createStopReplicaResponse()); for (AbstractRequestResponse req: requestResponseList) { ByteBuffer buffer = ByteBuffer.allocate(req.sizeOf()); @@ -184,4 +184,16 @@ private AbstractRequestResponse createProduceResponse() { responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE.code(), 10000)); return new ProduceResponse(responseData); } + + private AbstractRequest createStopReplicaRequest() { + Set partitions = new HashSet<>(); + partitions.add(new TopicPartition("test", 0)); + return new StopReplicaRequest(0, 1, true, partitions); + } + + private AbstractRequestResponse createStopReplicaResponse() { + Map responses = new HashMap<>(); + responses.put(new TopicPartition("test", 0), Errors.NONE.code()); + return new StopReplicaResponse(responses, Errors.NONE.code()); + } } diff --git a/core/src/main/scala/kafka/api/RequestKeys.scala b/core/src/main/scala/kafka/api/RequestKeys.scala index 155cb650e9cff..e66f018d99419 100644 --- a/core/src/main/scala/kafka/api/RequestKeys.scala +++ b/core/src/main/scala/kafka/api/RequestKeys.scala @@ -43,7 +43,6 @@ object RequestKeys { OffsetsKey -> ("Offsets", OffsetRequest.readFrom), MetadataKey -> ("Metadata", TopicMetadataRequest.readFrom), LeaderAndIsrKey -> ("LeaderAndIsr", LeaderAndIsrRequest.readFrom), - StopReplicaKey -> ("StopReplica", StopReplicaRequest.readFrom), UpdateMetadataKey -> ("UpdateMetadata", UpdateMetadataRequest.readFrom), ControlledShutdownKey -> ("ControlledShutdown", ControlledShutdownRequest.readFrom), OffsetCommitKey -> ("OffsetCommit", OffsetCommitRequest.readFrom), diff --git a/core/src/main/scala/kafka/api/StopReplicaRequest.scala b/core/src/main/scala/kafka/api/StopReplicaRequest.scala deleted file mode 100644 index 4441fc677ca48..0000000000000 --- a/core/src/main/scala/kafka/api/StopReplicaRequest.scala +++ /dev/null @@ -1,125 +0,0 @@ -/** - * 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.api - -import java.nio._ -import kafka.api.ApiUtils._ -import kafka.network.{RequestOrResponseSend, RequestChannel, InvalidRequestException} -import kafka.common.{TopicAndPartition, ErrorMapping} -import kafka.network.RequestChannel.Response -import kafka.utils.Logging -import collection.Set - - -object StopReplicaRequest extends Logging { - val CurrentVersion = 0.shortValue - val DefaultClientId = "" - val DefaultAckTimeout = 100 - - def readFrom(buffer: ByteBuffer): StopReplicaRequest = { - val versionId = buffer.getShort - val correlationId = buffer.getInt - val clientId = readShortString(buffer) - val controllerId = buffer.getInt - val controllerEpoch = buffer.getInt - val deletePartitions = buffer.get match { - case 1 => true - case 0 => false - case x => - throw new InvalidRequestException("Invalid byte %d in delete partitions field. (Assuming false.)".format(x)) - } - val topicPartitionPairCount = buffer.getInt - val topicPartitionPairSet = new collection.mutable.HashSet[TopicAndPartition]() - (1 to topicPartitionPairCount) foreach { _ => - topicPartitionPairSet.add(TopicAndPartition(readShortString(buffer), buffer.getInt)) - } - StopReplicaRequest(versionId, correlationId, clientId, controllerId, controllerEpoch, - deletePartitions, topicPartitionPairSet.toSet) - } -} - -case class StopReplicaRequest(versionId: Short, - correlationId: Int, - clientId: String, - controllerId: Int, - controllerEpoch: Int, - deletePartitions: Boolean, - partitions: Set[TopicAndPartition]) - extends RequestOrResponse(Some(RequestKeys.StopReplicaKey)) { - - def this(deletePartitions: Boolean, partitions: Set[TopicAndPartition], controllerId: Int, controllerEpoch: Int, correlationId: Int) = { - this(StopReplicaRequest.CurrentVersion, correlationId, StopReplicaRequest.DefaultClientId, - controllerId, controllerEpoch, deletePartitions, partitions) - } - - def writeTo(buffer: ByteBuffer) { - buffer.putShort(versionId) - buffer.putInt(correlationId) - writeShortString(buffer, clientId) - buffer.putInt(controllerId) - buffer.putInt(controllerEpoch) - buffer.put(if (deletePartitions) 1.toByte else 0.toByte) - buffer.putInt(partitions.size) - for (topicAndPartition <- partitions) { - writeShortString(buffer, topicAndPartition.topic) - buffer.putInt(topicAndPartition.partition) - } - } - - def sizeInBytes(): Int = { - var size = - 2 + /* versionId */ - 4 + /* correlation id */ - ApiUtils.shortStringLength(clientId) + - 4 + /* controller id*/ - 4 + /* controller epoch */ - 1 + /* deletePartitions */ - 4 /* partition count */ - for (topicAndPartition <- partitions){ - size += (ApiUtils.shortStringLength(topicAndPartition.topic)) + - 4 /* partition id */ - } - size - } - - override def toString(): String = { - describe(true) - } - - override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = { - val responseMap = partitions.map { - case topicAndPartition => (topicAndPartition, ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]])) - }.toMap - val errorResponse = StopReplicaResponse(correlationId, responseMap) - requestChannel.sendResponse(new Response(request, new RequestOrResponseSend(request.connectionId, errorResponse))) - } - - override def describe(details: Boolean): String = { - val stopReplicaRequest = new StringBuilder - stopReplicaRequest.append("Name: " + this.getClass.getSimpleName) - stopReplicaRequest.append("; Version: " + versionId) - stopReplicaRequest.append("; CorrelationId: " + correlationId) - stopReplicaRequest.append("; ClientId: " + clientId) - stopReplicaRequest.append("; DeletePartitions: " + deletePartitions) - stopReplicaRequest.append("; ControllerId: " + controllerId) - stopReplicaRequest.append("; ControllerEpoch: " + controllerEpoch) - if(details) - stopReplicaRequest.append("; Partitions: " + partitions.mkString(",")) - stopReplicaRequest.toString() - } -} \ No newline at end of file diff --git a/core/src/main/scala/kafka/api/StopReplicaResponse.scala b/core/src/main/scala/kafka/api/StopReplicaResponse.scala deleted file mode 100644 index 2fc3c9585fbc6..0000000000000 --- a/core/src/main/scala/kafka/api/StopReplicaResponse.scala +++ /dev/null @@ -1,75 +0,0 @@ -/** - * 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.api - -import java.nio.ByteBuffer -import collection.mutable.HashMap -import collection.immutable.Map -import kafka.common.{TopicAndPartition, ErrorMapping} -import kafka.api.ApiUtils._ - - -object StopReplicaResponse { - def readFrom(buffer: ByteBuffer): StopReplicaResponse = { - val correlationId = buffer.getInt - val errorCode = buffer.getShort - val numEntries = buffer.getInt - - val responseMap = new HashMap[TopicAndPartition, Short]() - for (i<- 0 until numEntries){ - val topic = readShortString(buffer) - val partition = buffer.getInt - val partitionErrorCode = buffer.getShort() - responseMap.put(TopicAndPartition(topic, partition), partitionErrorCode) - } - new StopReplicaResponse(correlationId, responseMap.toMap, errorCode) - } -} - - -case class StopReplicaResponse(correlationId: Int, - responseMap: Map[TopicAndPartition, Short], - errorCode: Short = ErrorMapping.NoError) - extends RequestOrResponse() { - def sizeInBytes(): Int ={ - var size = - 4 /* correlation id */ + - 2 /* error code */ + - 4 /* number of responses */ - for ((key, value) <- responseMap) { - size += - 2 + key.topic.length /* topic */ + - 4 /* partition */ + - 2 /* error code for this partition */ - } - size - } - - def writeTo(buffer: ByteBuffer) { - buffer.putInt(correlationId) - buffer.putShort(errorCode) - buffer.putInt(responseMap.size) - for ((topicAndPartition, errorCode) <- responseMap){ - writeShortString(buffer, topicAndPartition.topic) - buffer.putInt(topicAndPartition.partition) - buffer.putShort(errorCode) - } - } - - override def describe(details: Boolean):String = { toString } -} diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala index 4396b6e12d988..735a0298f5300 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -18,16 +18,29 @@ package kafka.controller import kafka.network.BlockingChannel import kafka.utils.{CoreUtils, Logging, ShutdownableThread} +import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.network.NetworkReceive +import org.apache.kafka.common.protocol.ApiKeys +import org.apache.kafka.common.requests.{ResponseHeader, RequestHeader, AbstractRequestResponse, AbstractRequest, StopReplicaRequest, StopReplicaResponse} import collection.mutable.HashMap import kafka.cluster.Broker import java.util.concurrent.{LinkedBlockingQueue, BlockingQueue} import kafka.server.KafkaConfig import collection.mutable import kafka.api._ -import kafka.common.TopicAndPartition +import kafka.common.{KafkaException, TopicAndPartition} import kafka.api.RequestOrResponse import collection.Set +import scala.collection.JavaConverters._ + +sealed trait RequestQueued + +case class ScalaRequestQueued(request: RequestOrResponse, + callback: (RequestOrResponse) => Unit) extends RequestQueued + +case class JavaRequestQueued(header: RequestHeader, + body: AbstractRequest, + callback: (ResponseHeader, AbstractRequestResponse) => Unit) extends RequestQueued class ControllerChannelManager (private val controllerContext: ControllerContext, config: KafkaConfig) extends Logging { protected val brokerStateInfo = new HashMap[Int, ControllerBrokerStateInfo] @@ -48,12 +61,25 @@ class ControllerChannelManager (private val controllerContext: ControllerContext } } + def sendRequest(brokerId: Int, header: RequestHeader, body: AbstractRequest, callback: (ResponseHeader, AbstractRequestResponse) => Unit): Unit = { + brokerLock synchronized { + val stateInfoOpt = brokerStateInfo.get(brokerId) + stateInfoOpt match { + case Some(stateInfo) => + stateInfo.messageQueue.put(JavaRequestQueued(header, body, callback)) + case None => + warn("Not sending request %s to broker %d, since it is offline.".format(body, brokerId)) + } + } + } + + // TODO Remove this one when migration of all requests / responses to org.apache.kafka.common.requests is done. def sendRequest(brokerId : Int, request : RequestOrResponse, callback: (RequestOrResponse) => Unit = null) { brokerLock synchronized { val stateInfoOpt = brokerStateInfo.get(brokerId) stateInfoOpt match { case Some(stateInfo) => - stateInfo.messageQueue.put((request, callback)) + stateInfo.messageQueue.put(ScalaRequestQueued(request, callback)) case None => warn("Not sending request %s to broker %d, since it is offline.".format(request, brokerId)) } @@ -77,7 +103,7 @@ class ControllerChannelManager (private val controllerContext: ControllerContext } private def addNewBroker(broker: Broker) { - val messageQueue = new LinkedBlockingQueue[(RequestOrResponse, (RequestOrResponse) => Unit)]() + val messageQueue = new LinkedBlockingQueue[RequestQueued]() debug("Controller %d trying to connect to broker %d".format(config.brokerId,broker.id)) val brokerEndPoint = broker.getBrokerEndPoint(config.interBrokerSecurityProtocol) val channel = new BlockingChannel(brokerEndPoint.host, brokerEndPoint.port, @@ -110,7 +136,7 @@ class ControllerChannelManager (private val controllerContext: ControllerContext class RequestSendThread(val controllerId: Int, val controllerContext: ControllerContext, val toBroker: Broker, - val queue: BlockingQueue[(RequestOrResponse, (RequestOrResponse) => Unit)], + val queue: BlockingQueue[RequestQueued], val channel: BlockingChannel) extends ShutdownableThread("Controller-%d-to-broker-%d-send-thread".format(controllerId, toBroker.id)) { private val lock = new Object() @@ -119,47 +145,11 @@ class RequestSendThread(val controllerId: Int, override def doWork(): Unit = { val queueItem = queue.take() - val request = queueItem._1 - val callback = queueItem._2 - var receive: NetworkReceive = null try { lock synchronized { - var isSendSuccessful = false - while (isRunning.get() && !isSendSuccessful) { - // if a broker goes down for a long time, then at some point the controller's zookeeper listener will trigger a - // removeBroker which will invoke shutdown() on this thread. At that point, we will stop retrying. - try { - channel.send(request) - receive = channel.receive() - isSendSuccessful = true - } catch { - case e: Throwable => // if the send was not successful, reconnect to broker and resend the message - warn(("Controller %d epoch %d fails to send request %s to broker %s. " + - "Reconnecting to broker.").format(controllerId, controllerContext.epoch, - request.toString, toBroker.toString()), e) - channel.disconnect() - connectToBroker(toBroker, channel) - isSendSuccessful = false - // backoff before retrying the connection and send - CoreUtils.swallowTrace(Thread.sleep(300)) - } - } - if (receive != null) { - var response: RequestOrResponse = null - request.requestId.get match { - case RequestKeys.LeaderAndIsrKey => - response = LeaderAndIsrResponse.readFrom(receive.payload()) - case RequestKeys.StopReplicaKey => - response = StopReplicaResponse.readFrom(receive.payload()) - case RequestKeys.UpdateMetadataKey => - response = UpdateMetadataResponse.readFrom(receive.payload()) - } - stateChangeLogger.trace("Controller %d epoch %d received response %s for a request sent to broker %s" - .format(controllerId, controllerContext.epoch, response.toString, toBroker.toString)) - - if (callback != null) { - callback(response) - } + queueItem match { + case ScalaRequestQueued(request, callback) => sendScalaRequest(request, callback) + case JavaRequestQueued(header, body, callback) => sendJavaRequest(header, body, callback) } } } catch { @@ -170,6 +160,89 @@ class RequestSendThread(val controllerId: Int, } } + private def sendJavaRequest(header: RequestHeader, body: AbstractRequest, + callback: (ResponseHeader, AbstractRequestResponse) => Unit): Unit = { + var receive: NetworkReceive = null + var isSendSuccessful = false + + while (isRunning.get() && !isSendSuccessful) { + // if a broker goes down for a long time, then at some point the controller's zookeeper listener will trigger a + // removeBroker which will invoke shutdown() on this thread. At that point, we will stop retrying. + try { + channel.send(header, body) + receive = channel.receive() + isSendSuccessful = true + } catch { + case e: Throwable => // if the send was not successful, reconnect to broker and resend the message + warn(("Controller %d epoch %d fails to send request %s to broker %s. " + + "Reconnecting to broker.").format(controllerId, controllerContext.epoch, + body.toString, toBroker.toString()), e) + channel.disconnect() + connectToBroker(toBroker, channel) + isSendSuccessful = false + // backoff before retrying the connection and send + CoreUtils.swallowTrace(Thread.sleep(300)) + } + } + + if (receive != null) { + val rspHeader = ResponseHeader.parse(receive.payload()) + val apiKey = ApiKeys.forId(header.apiKey()) + val rspBody = apiKey match { + case ApiKeys.STOP_REPLICA => StopReplicaResponse.parse(receive.payload()) + case _ => throw new KafkaException("Unknown api code " + apiKey.id) + } + stateChangeLogger.trace("Controller %d epoch %d received response %s for a request sent to broker %s" + .format(controllerId, controllerContext.epoch, rspBody.toString, toBroker.toString)) + + if (callback != null) { + callback(rspHeader, rspBody) + } + } + } + + // TODO Remove this one when migration of all requests / responses to org.apache.kafka.common.requests is done. + private def sendScalaRequest(request: RequestOrResponse, callback: (RequestOrResponse) => Unit): Unit = { + var receive: NetworkReceive = null + var isSendSuccessful = false + + while (isRunning.get() && !isSendSuccessful) { + // if a broker goes down for a long time, then at some point the controller's zookeeper listener will trigger a + // removeBroker which will invoke shutdown() on this thread. At that point, we will stop retrying. + try { + channel.send(request) + receive = channel.receive() + isSendSuccessful = true + } catch { + case e: Throwable => // if the send was not successful, reconnect to broker and resend the message + warn(("Controller %d epoch %d fails to send request %s to broker %s. " + + "Reconnecting to broker.").format(controllerId, controllerContext.epoch, + request.toString, toBroker.toString()), e) + channel.disconnect() + connectToBroker(toBroker, channel) + isSendSuccessful = false + // backoff before retrying the connection and send + CoreUtils.swallowTrace(Thread.sleep(300)) + } + } + + if (receive != null) { + var response: RequestOrResponse = null + request.requestId.get match { + case RequestKeys.LeaderAndIsrKey => + response = LeaderAndIsrResponse.readFrom(receive.payload()) + case RequestKeys.UpdateMetadataKey => + response = UpdateMetadataResponse.readFrom(receive.payload()) + } + stateChangeLogger.trace("Controller %d epoch %d received response %s for a request sent to broker %s" + .format(controllerId, controllerContext.epoch, response.toString, toBroker.toString)) + + if (callback != null) { + callback(response) + } + } + } + private def connectToBroker(broker: Broker, channel: BlockingChannel) { try { channel.connect() @@ -222,13 +295,13 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends Logging } def addStopReplicaRequestForBrokers(brokerIds: Seq[Int], topic: String, partition: Int, deletePartition: Boolean, - callback: (RequestOrResponse, Int) => Unit = null) { + callback: (ResponseHeader, StopReplicaResponse, Int) => Unit = null) { brokerIds.filter(b => b >= 0).foreach { brokerId => stopReplicaRequestMap.getOrElseUpdate(brokerId, Seq.empty[StopReplicaRequestInfo]) val v = stopReplicaRequestMap(brokerId) if(callback != null) stopReplicaRequestMap(brokerId) = v :+ StopReplicaRequestInfo(PartitionAndReplica(topic, partition, brokerId), - deletePartition, (r: RequestOrResponse) => { callback(r, brokerId) }) + deletePartition, (header, body) => { callback(header, body.asInstanceOf[StopReplicaResponse], brokerId) }) else stopReplicaRequestMap(brokerId) = v :+ StopReplicaRequestInfo(PartitionAndReplica(topic, partition, brokerId), deletePartition) @@ -318,9 +391,10 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends Logging debug("The stop replica request (delete = false) sent to broker %d is %s" .format(broker, stopReplicaWithoutDelete.mkString(","))) replicaInfoList.foreach { r => - val stopReplicaRequest = new StopReplicaRequest(r.deletePartition, - Set(TopicAndPartition(r.replica.topic, r.replica.partition)), controllerId, controllerEpoch, correlationId) - controller.sendRequest(broker, stopReplicaRequest, r.callback) + val stopReplicaHeader = new RequestHeader(ApiKeys.STOP_REPLICA.id, "", correlationId) + val stopReplicaRequest = new StopReplicaRequest(controllerId, controllerEpoch, r.deletePartition, + Set(new TopicPartition(r.replica.topic, r.replica.partition)).asJava) + controller.sendRequest(broker, stopReplicaHeader, stopReplicaRequest, r.callback) } } stopReplicaRequestMap.clear() @@ -346,20 +420,22 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends Logging case class ControllerBrokerStateInfo(channel: BlockingChannel, broker: Broker, - messageQueue: BlockingQueue[(RequestOrResponse, (RequestOrResponse) => Unit)], + messageQueue: BlockingQueue[RequestQueued], requestSendThread: RequestSendThread) -case class StopReplicaRequestInfo(replica: PartitionAndReplica, deletePartition: Boolean, callback: (RequestOrResponse) => Unit = null) +case class StopReplicaRequestInfo(replica: PartitionAndReplica, + deletePartition: Boolean, + callback: (ResponseHeader, AbstractRequestResponse) => Unit = null) class Callbacks private (var leaderAndIsrResponseCallback:(RequestOrResponse) => Unit = null, var updateMetadataResponseCallback:(RequestOrResponse) => Unit = null, - var stopReplicaResponseCallback:(RequestOrResponse, Int) => Unit = null) + var stopReplicaResponseCallback:(ResponseHeader, StopReplicaResponse, Int) => Unit = null) object Callbacks { class CallbackBuilder { var leaderAndIsrResponseCbk:(RequestOrResponse) => Unit = null var updateMetadataResponseCbk:(RequestOrResponse) => Unit = null - var stopReplicaResponseCbk:(RequestOrResponse, Int) => Unit = null + var stopReplicaResponseCbk:(ResponseHeader, StopReplicaResponse, Int) => Unit = null def leaderAndIsrCallback(cbk: (RequestOrResponse) => Unit): CallbackBuilder = { leaderAndIsrResponseCbk = cbk @@ -371,7 +447,7 @@ object Callbacks { this } - def stopReplicaCallback(cbk: (RequestOrResponse, Int) => Unit): CallbackBuilder = { + def stopReplicaCallback(cbk: (ResponseHeader, StopReplicaResponse, Int) => Unit): CallbackBuilder = { stopReplicaResponseCbk = cbk this } diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index 4c376168c1af4..6f5b9c43e53ef 100755 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/core/src/main/scala/kafka/controller/KafkaController.scala @@ -18,6 +18,8 @@ package kafka.controller import java.util +import org.apache.kafka.common.requests.{AbstractRequestResponse, ResponseHeader, AbstractRequest, RequestHeader} + import scala.collection._ import com.yammer.metrics.core.Gauge import java.util.concurrent.TimeUnit @@ -690,6 +692,13 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient, val brokerSt onControllerResignation() } + def sendRequest(brokerId: Int, + header: RequestHeader, + body: AbstractRequest, + callback: (ResponseHeader, AbstractRequestResponse) => Unit) = { + controllerContext.controllerChannelManager.sendRequest(brokerId, header, body, callback) + } + def sendRequest(brokerId : Int, request : RequestOrResponse, callback: (RequestOrResponse) => Unit = null) = { controllerContext.controllerChannelManager.sendRequest(brokerId, request, callback) } diff --git a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala index 64b11df5e0056..14e06b44f2361 100755 --- a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala +++ b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala @@ -24,9 +24,10 @@ import kafka.utils.{ShutdownableThread, Logging, ZkUtils} import kafka.utils.CoreUtils._ import collection.Set import kafka.common.{ErrorMapping, TopicAndPartition} -import kafka.api.{StopReplicaResponse, RequestOrResponse} +import org.apache.kafka.common.requests.{ResponseHeader, StopReplicaResponse} import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.atomic.AtomicBoolean +import scala.collection.JavaConverters._ /** * This manages the state machine for topic deletion. @@ -363,21 +364,21 @@ class TopicDeletionManager(controller: KafkaController, startReplicaDeletion(replicasPerPartition) } - private def deleteTopicStopReplicaCallback(stopReplicaResponseObj: RequestOrResponse, replicaId: Int) { - val stopReplicaResponse = stopReplicaResponseObj.asInstanceOf[StopReplicaResponse] + private def deleteTopicStopReplicaCallback(responseHeader: ResponseHeader, stopReplicaResponse: StopReplicaResponse, replicaId: Int) { debug("Delete topic callback invoked for %s".format(stopReplicaResponse)) - val partitionsInError = if(stopReplicaResponse.errorCode != ErrorMapping.NoError) { - stopReplicaResponse.responseMap.keySet + val responses = stopReplicaResponse.responses().asScala + val partitionsInError = if(stopReplicaResponse.errorCode() != ErrorMapping.NoError) { + responses.keySet } else - stopReplicaResponse.responseMap.filter(p => p._2 != ErrorMapping.NoError).map(_._1).toSet - val replicasInError = partitionsInError.map(p => PartitionAndReplica(p.topic, p.partition, replicaId)) + responses.filter(p => p._2 != ErrorMapping.NoError).map(_._1).toSet + val replicasInError = partitionsInError.map(p => PartitionAndReplica(p.topic(), p.partition(), replicaId)) inLock(controllerContext.controllerLock) { // move all the failed replicas to ReplicaDeletionIneligible failReplicaDeletion(replicasInError) - if(replicasInError.size != stopReplicaResponse.responseMap.size) { + if(replicasInError.size != responses.size) { // some replicas could have been successfully deleted - val deletedReplicas = stopReplicaResponse.responseMap.keySet -- partitionsInError - completeReplicaDeletion(deletedReplicas.map(p => PartitionAndReplica(p.topic, p.partition, replicaId))) + val deletedReplicas = responses.keySet -- partitionsInError + completeReplicaDeletion(deletedReplicas.map(p => PartitionAndReplica(p.topic(), p.partition(), replicaId))) } } } diff --git a/core/src/main/scala/kafka/network/BlockingChannel.scala b/core/src/main/scala/kafka/network/BlockingChannel.scala index 1197259cd9f59..cd3f69ca10971 100644 --- a/core/src/main/scala/kafka/network/BlockingChannel.scala +++ b/core/src/main/scala/kafka/network/BlockingChannel.scala @@ -23,6 +23,7 @@ import java.nio.channels._ import kafka.api.RequestOrResponse import kafka.utils.{Logging, nonthreadsafe} import org.apache.kafka.common.network.NetworkReceive +import org.apache.kafka.common.requests.{RequestSend, AbstractRequest, RequestHeader} object BlockingChannel{ @@ -105,6 +106,14 @@ class BlockingChannel( val host: String, def isConnected = connected + def send(header: RequestHeader, body: AbstractRequest): Long = { + if(!connected) + throw new ClosedChannelException() + + val send = new RequestSend(connectionId, header, body.toStruct) + send.writeCompletelyTo(writeChannel) + } + def send(request: RequestOrResponse): Long = { if(!connected) throw new ClosedChannelException() diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 67f0cad802f90..9b2c57ee2438b 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -28,9 +28,10 @@ import kafka.coordinator.ConsumerCoordinator import kafka.log._ import kafka.network._ import kafka.network.RequestChannel.Response -import org.apache.kafka.common.requests.{JoinGroupRequest, JoinGroupResponse, HeartbeatRequest, HeartbeatResponse, ResponseHeader, ResponseSend} +import org.apache.kafka.common.requests.{JoinGroupRequest, JoinGroupResponse, HeartbeatRequest, HeartbeatResponse, ResponseHeader, ResponseSend, StopReplicaRequest, StopReplicaResponse} import kafka.utils.{ZkUtils, ZKGroupTopicDirs, SystemTime, Logging} import scala.collection._ +import scala.collection.JavaConverters._ import org.I0Itec.zkclient.ZkClient /** @@ -127,10 +128,12 @@ class KafkaApis(val requestChannel: RequestChannel, // ensureTopicExists is only for client facing requests // We can't have the ensureTopicExists check here since the controller sends it as an advisory to all brokers so they // stop serving data to clients for the topic being deleted - val stopReplicaRequest = request.requestObj.asInstanceOf[StopReplicaRequest] + val stopReplicaRequest = request.body.asInstanceOf[StopReplicaRequest] + val respHeader = new ResponseHeader(request.header.correlationId) val (response, error) = replicaManager.stopReplicas(stopReplicaRequest) - val stopReplicaResponse = new StopReplicaResponse(stopReplicaRequest.correlationId, response.toMap, error) - requestChannel.sendResponse(new Response(request, new RequestOrResponseSend(request.connectionId, stopReplicaResponse))) + val stopReplicaResponse = new StopReplicaResponse(response.asInstanceOf[Map[TopicPartition, java.lang.Short]].asJava, error) + + requestChannel.sendResponse(new RequestChannel.Response(request, new ResponseSend(request.connectionId, respHeader, stopReplicaResponse))) replicaManager.replicaFetcherManager.shutdownIdleFetcherThreads() } diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index d829e180c3943..c05bcdcb58e30 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -32,8 +32,11 @@ import kafka.utils._ import org.I0Itec.zkclient.ZkClient import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.metrics.Metrics +import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.requests.StopReplicaRequest import scala.collection._ +import scala.collection.JavaConverters._ /* * Result metadata of a log append operation on the log @@ -224,21 +227,22 @@ class ReplicaManager(val config: KafkaConfig, errorCode } - def stopReplicas(stopReplicaRequest: StopReplicaRequest): (mutable.Map[TopicAndPartition, Short], Short) = { + def stopReplicas(stopReplicaRequest: StopReplicaRequest): (mutable.Map[TopicPartition, Short], Short) = { replicaStateChangeLock synchronized { - val responseMap = new collection.mutable.HashMap[TopicAndPartition, Short] - if(stopReplicaRequest.controllerEpoch < controllerEpoch) { + val responseMap = new collection.mutable.HashMap[TopicPartition, Short] + if(stopReplicaRequest.controllerEpoch() < controllerEpoch) { stateChangeLogger.warn("Broker %d received stop replica request from an old controller epoch %d." - .format(localBrokerId, stopReplicaRequest.controllerEpoch) + + .format(localBrokerId, stopReplicaRequest.controllerEpoch()) + " Latest known controller epoch is %d " + controllerEpoch) (responseMap, ErrorMapping.StaleControllerEpochCode) } else { - controllerEpoch = stopReplicaRequest.controllerEpoch + val partitions = stopReplicaRequest.partitions().asScala + controllerEpoch = stopReplicaRequest.controllerEpoch() // First stop fetchers for all partitions, then stop the corresponding replicas - replicaFetcherManager.removeFetcherForPartitions(stopReplicaRequest.partitions.map(r => TopicAndPartition(r.topic, r.partition))) - for(topicAndPartition <- stopReplicaRequest.partitions){ - val errorCode = stopReplica(topicAndPartition.topic, topicAndPartition.partition, stopReplicaRequest.deletePartitions) - responseMap.put(topicAndPartition, errorCode) + replicaFetcherManager.removeFetcherForPartitions(partitions.map(r => TopicAndPartition(r.topic, r.partition))) + for(topicPartition <- partitions){ + val errorCode = stopReplica(topicPartition.topic, topicPartition.partition, stopReplicaRequest.deletePartitions) + responseMap.put(topicPartition, errorCode) } (responseMap, ErrorMapping.NoError) } diff --git a/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala b/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala index b4c2a228c3c98..ab5990750b9ab 100644 --- a/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala +++ b/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala @@ -131,17 +131,6 @@ object SerializationTestUtils { new LeaderAndIsrResponse(1, responseMap) } - def createTestStopReplicaRequest() : StopReplicaRequest = { - new StopReplicaRequest(controllerId = 0, controllerEpoch = 1, correlationId = 0, deletePartitions = true, - partitions = collection.immutable.Set(TopicAndPartition(topic1, 0),TopicAndPartition(topic2, 0))) - } - - def createTestStopReplicaResponse() : StopReplicaResponse = { - val responseMap = Map((TopicAndPartition(topic1, 0), ErrorMapping.NoError), - (TopicAndPartition(topic2, 0), ErrorMapping.NoError)) - new StopReplicaResponse(0, responseMap.toMap) - } - def createTestProducerRequest: ProducerRequest = { new ProducerRequest(1, "client 1", 0, 1000, topicDataProducerRequest) } @@ -256,8 +245,6 @@ object SerializationTestUtils { class RequestResponseSerializationTest extends JUnitSuite { private val leaderAndIsrRequest = SerializationTestUtils.createTestLeaderAndIsrRequest private val leaderAndIsrResponse = SerializationTestUtils.createTestLeaderAndIsrResponse - private val stopReplicaRequest = SerializationTestUtils.createTestStopReplicaRequest - private val stopReplicaResponse = SerializationTestUtils.createTestStopReplicaResponse private val producerRequest = SerializationTestUtils.createTestProducerRequest private val producerResponse = SerializationTestUtils.createTestProducerResponse private val fetchRequest = SerializationTestUtils.createTestFetchRequest @@ -283,8 +270,8 @@ class RequestResponseSerializationTest extends JUnitSuite { def testSerializationAndDeserialization() { val requestsAndResponses = - collection.immutable.Seq(leaderAndIsrRequest, leaderAndIsrResponse, stopReplicaRequest, - stopReplicaResponse, producerRequest, producerResponse, + collection.immutable.Seq(leaderAndIsrRequest, leaderAndIsrResponse, + producerRequest, producerResponse, fetchRequest, offsetRequest, offsetResponse, topicMetadataRequest, topicMetadataResponse, offsetCommitRequestV0, offsetCommitRequestV1, offsetCommitRequestV2, diff --git a/core/src/test/scala/unit/kafka/controller/ControllerFailoverTest.scala b/core/src/test/scala/unit/kafka/controller/ControllerFailoverTest.scala index 0e38a18082450..761576a549219 100644 --- a/core/src/test/scala/unit/kafka/controller/ControllerFailoverTest.scala +++ b/core/src/test/scala/unit/kafka/controller/ControllerFailoverTest.scala @@ -168,7 +168,7 @@ class MockChannelManager(private val controllerContext: ControllerContext, } def shrinkBlockingQueue(brokerId: Int) { - val messageQueue = new LinkedBlockingQueue[(RequestOrResponse, RequestOrResponse => Unit)](1) + val messageQueue = new LinkedBlockingQueue[RequestQueued](1) val brokerInfo = this.brokerStateInfo(brokerId) this.brokerStateInfo.put(brokerId, new ControllerBrokerStateInfo(brokerInfo.channel, brokerInfo.broker, @@ -187,4 +187,4 @@ class MockChannelManager(private val controllerContext: ControllerContext, def queueSize(brokerId: Int): Int = { this.brokerStateInfo(brokerId).messageQueue.size } -} \ No newline at end of file +}