You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2018/08/24 15:32:19 UTC
[kafka] branch trunk updated: KAFKA-5975;
No response when deleting topics and delete.topic.enable=false
(#3960)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0477091 KAFKA-5975; No response when deleting topics and delete.topic.enable=false (#3960)
0477091 is described below
commit 04770916a743ca9526cfa2fdecaef419d429a443
Author: Mario Molina <mm...@gmail.com>
AuthorDate: Fri Aug 24 10:32:06 2018 -0500
KAFKA-5975; No response when deleting topics and delete.topic.enable=false (#3960)
This PR implements [KIP-322](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=87295558).
Reviewers: Tom Bentley <tb...@redhat.com>, Manikumar Reddy O <ma...@gmail.com>, Jason Gustafson <ja...@confluent.io>
---
.../errors/TopicDeletionDisabledException.java | 29 +++++++++++
.../org/apache/kafka/common/protocol/Errors.java | 41 ++++++++-------
.../kafka/common/requests/DeleteTopicsRequest.java | 9 +++-
.../common/requests/DeleteTopicsResponse.java | 11 +++-
.../kafka/clients/admin/KafkaAdminClientTest.java | 29 +++++++++++
core/src/main/scala/kafka/server/KafkaApis.scala | 6 +++
...leteTopicsRequestWithDeletionDisabledTest.scala | 59 ++++++++++++++++++++++
7 files changed, 163 insertions(+), 21 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/TopicDeletionDisabledException.java b/clients/src/main/java/org/apache/kafka/common/errors/TopicDeletionDisabledException.java
new file mode 100644
index 0000000..41577d2
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/TopicDeletionDisabledException.java
@@ -0,0 +1,29 @@
+/*
+ * 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 TopicDeletionDisabledException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public TopicDeletionDisabledException() {
+ }
+
+ public TopicDeletionDisabledException(String message) {
+ super(message);
+ }
+}
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 090dca3..d461060 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
@@ -80,6 +80,7 @@ import org.apache.kafka.common.errors.SaslAuthenticationException;
import org.apache.kafka.common.errors.SecurityDisabledException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.common.errors.TopicDeletionDisabledException;
import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
import org.apache.kafka.common.errors.TransactionCoordinatorFencedException;
@@ -110,7 +111,7 @@ import java.util.function.Function;
* Do not add exceptions that occur only on the client or only on the server here.
*/
public enum Errors {
- UNKNOWN_SERVER_ERROR(-1, "The server experienced an unexpected error when processing the request",
+ UNKNOWN_SERVER_ERROR(-1, "The server experienced an unexpected error when processing the request,",
UnknownServerException::new),
NONE(0, null, message -> null),
OFFSET_OUT_OF_RANGE(1, "The requested offset is not within the range of offsets maintained by the server.",
@@ -129,7 +130,7 @@ public enum Errors {
TimeoutException::new),
BROKER_NOT_AVAILABLE(8, "The broker is not available.",
BrokerNotAvailableException::new),
- REPLICA_NOT_AVAILABLE(9, "The replica is not available for the requested topic-partition",
+ REPLICA_NOT_AVAILABLE(9, "The replica is not available for the requested topic-partition.",
ReplicaNotAvailableException::new),
MESSAGE_TOO_LARGE(10, "The request included a message larger than the max message size the server will accept.",
RecordTooLargeException::new),
@@ -161,7 +162,7 @@ public enum Errors {
"The group member's supported protocols are incompatible with those of existing members" +
" or first group member tried to join with empty protocol type or empty protocol list.",
InconsistentGroupProtocolException::new),
- INVALID_GROUP_ID(24, "The configured groupId is invalid",
+ INVALID_GROUP_ID(24, "The configured groupId is invalid.",
InvalidGroupIdException::new),
UNKNOWN_MEMBER_ID(25, "The coordinator is not aware of this member.",
UnknownMemberIdException::new),
@@ -171,7 +172,7 @@ public enum Errors {
InvalidSessionTimeoutException::new),
REBALANCE_IN_PROGRESS(27, "The group is rebalancing, so a rejoin is needed.",
RebalanceInProgressException::new),
- INVALID_COMMIT_OFFSET_SIZE(28, "The committing offset data size is not valid",
+ INVALID_COMMIT_OFFSET_SIZE(28, "The committing offset data size is not valid.",
InvalidCommitOffsetSizeException::new),
TOPIC_AUTHORIZATION_FAILED(29, "Topic authorization failed.",
TopicAuthorizationException::new),
@@ -207,28 +208,28 @@ public enum Errors {
UnsupportedForMessageFormatException::new),
POLICY_VIOLATION(44, "Request parameters do not satisfy the configured policy.",
PolicyViolationException::new),
- OUT_OF_ORDER_SEQUENCE_NUMBER(45, "The broker received an out of order sequence number",
+ OUT_OF_ORDER_SEQUENCE_NUMBER(45, "The broker received an out of order sequence number.",
OutOfOrderSequenceException::new),
- DUPLICATE_SEQUENCE_NUMBER(46, "The broker received a duplicate sequence number",
+ DUPLICATE_SEQUENCE_NUMBER(46, "The broker received a duplicate sequence number.",
DuplicateSequenceException::new),
INVALID_PRODUCER_EPOCH(47, "Producer attempted an operation with an old epoch. Either there is a newer producer " +
"with the same transactionalId, or the producer's transaction has been expired by the broker.",
ProducerFencedException::new),
- INVALID_TXN_STATE(48, "The producer attempted a transactional operation in an invalid state",
+ INVALID_TXN_STATE(48, "The producer attempted a transactional operation in an invalid state.",
InvalidTxnStateException::new),
INVALID_PRODUCER_ID_MAPPING(49, "The producer attempted to use a producer id which is not currently assigned to " +
- "its transactional id",
+ "its transactional id.",
InvalidPidMappingException::new),
INVALID_TRANSACTION_TIMEOUT(50, "The transaction timeout is larger than the maximum value allowed by " +
"the broker (as configured by transaction.max.timeout.ms).",
InvalidTxnTimeoutException::new),
CONCURRENT_TRANSACTIONS(51, "The producer attempted to update a transaction " +
- "while another concurrent operation on the same transaction was ongoing",
+ "while another concurrent operation on the same transaction was ongoing.",
ConcurrentTransactionsException::new),
TRANSACTION_COORDINATOR_FENCED(52, "Indicates that the transaction coordinator sending a WriteTxnMarker " +
- "is no longer the current coordinator for a given producer",
+ "is no longer the current coordinator for a given producer.",
TransactionCoordinatorFencedException::new),
- TRANSACTIONAL_ID_AUTHORIZATION_FAILED(53, "Transactional Id authorization failed",
+ TRANSACTIONAL_ID_AUTHORIZATION_FAILED(53, "Transactional Id authorization failed.",
TransactionalIdAuthorizationException::new),
SECURITY_DISABLED(54, "Security features are disabled.",
SecurityDisabledException::new),
@@ -248,7 +249,7 @@ public enum Errors {
"removed, the producer's metadata is removed from the broker, and future appends by the producer will " +
"return this exception.",
UnknownProducerIdException::new),
- REASSIGNMENT_IN_PROGRESS(60, "A partition reassignment is in progress",
+ REASSIGNMENT_IN_PROGRESS(60, "A partition reassignment is in progress.",
ReassignmentInProgressException::new),
DELEGATION_TOKEN_AUTH_DISABLED(61, "Delegation Token feature is not enabled.",
DelegationTokenDisabledException::new),
@@ -263,19 +264,21 @@ public enum Errors {
DelegationTokenAuthorizationException::new),
DELEGATION_TOKEN_EXPIRED(66, "Delegation Token is expired.",
DelegationTokenExpiredException::new),
- INVALID_PRINCIPAL_TYPE(67, "Supplied principalType is not supported",
+ INVALID_PRINCIPAL_TYPE(67, "Supplied principalType is not supported.",
InvalidPrincipalTypeException::new),
- NON_EMPTY_GROUP(68, "The group is not empty",
+ NON_EMPTY_GROUP(68, "The group is not empty.",
GroupNotEmptyException::new),
- GROUP_ID_NOT_FOUND(69, "The group id does not exist",
+ GROUP_ID_NOT_FOUND(69, "The group id does not exist.",
GroupIdNotFoundException::new),
- FETCH_SESSION_ID_NOT_FOUND(70, "The fetch session ID was not found",
+ FETCH_SESSION_ID_NOT_FOUND(70, "The fetch session ID was not found.",
FetchSessionIdNotFoundException::new),
- INVALID_FETCH_SESSION_EPOCH(71, "The fetch session epoch is invalid",
+ INVALID_FETCH_SESSION_EPOCH(71, "The fetch session epoch is invalid.",
InvalidFetchSessionEpochException::new),
LISTENER_NOT_FOUND(72, "There is no listener on the leader broker that matches the listener on which " +
- "metadata request was processed",
- ListenerNotFoundException::new),;
+ "metadata request was processed.",
+ ListenerNotFoundException::new),
+ TOPIC_DELETION_DISABLED(73, "Topic deletion is disabled.",
+ TopicDeletionDisabledException::new);
private static final Logger log = LoggerFactory.getLogger(Errors.class);
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
index dbcc25d..87f14b4 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
@@ -51,9 +51,15 @@ public class DeleteTopicsRequest extends AbstractRequest {
*/
private static final Schema DELETE_TOPICS_REQUEST_V2 = DELETE_TOPICS_REQUEST_V1;
+ /**
+ * v3 request is the same that as v2. The response is different based on the request version.
+ * In v3 version a TopicDeletionDisabledException is returned
+ */
+ private static final Schema DELETE_TOPICS_REQUEST_V3 = DELETE_TOPICS_REQUEST_V2;
+
public static Schema[] schemaVersions() {
return new Schema[]{DELETE_TOPICS_REQUEST_V0, DELETE_TOPICS_REQUEST_V1,
- DELETE_TOPICS_REQUEST_V2};
+ DELETE_TOPICS_REQUEST_V2, DELETE_TOPICS_REQUEST_V3};
}
private final Set<String> topics;
@@ -121,6 +127,7 @@ public class DeleteTopicsRequest extends AbstractRequest {
return new DeleteTopicsResponse(topicErrors);
case 1:
case 2:
+ case 3:
return new DeleteTopicsResponse(throttleTimeMs, topicErrors);
default:
throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
index db1c434..650caa8 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
@@ -53,8 +53,15 @@ public class DeleteTopicsResponse extends AbstractResponse {
*/
private static final Schema DELETE_TOPICS_RESPONSE_V2 = DELETE_TOPICS_RESPONSE_V1;
+ /**
+ * v3 request is the same that as v2. The response is different based on the request version.
+ * In v3 version a TopicDeletionDisabledException is returned
+ */
+ private static final Schema DELETE_TOPICS_RESPONSE_V3 = DELETE_TOPICS_RESPONSE_V2;
+
public static Schema[] schemaVersions() {
- return new Schema[]{DELETE_TOPICS_RESPONSE_V0, DELETE_TOPICS_RESPONSE_V1, DELETE_TOPICS_RESPONSE_V2};
+ return new Schema[]{DELETE_TOPICS_RESPONSE_V0, DELETE_TOPICS_RESPONSE_V1,
+ DELETE_TOPICS_RESPONSE_V2, DELETE_TOPICS_RESPONSE_V3};
}
@@ -65,6 +72,8 @@ public class DeleteTopicsResponse extends AbstractResponse {
* INVALID_TOPIC_EXCEPTION(17)
* TOPIC_AUTHORIZATION_FAILED(29)
* NOT_CONTROLLER(41)
+ * INVALID_REQUEST(42)
+ * TOPIC_DELETION_DISABLED(73)
*/
private final Map<String, Errors> errors;
private final int throttleTimeMs;
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 7079471..0245cbd 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -45,6 +45,7 @@ import org.apache.kafka.common.errors.OffsetOutOfRangeException;
import org.apache.kafka.common.errors.SaslAuthenticationException;
import org.apache.kafka.common.errors.SecurityDisabledException;
import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.TopicDeletionDisabledException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.ApiError;
@@ -58,6 +59,8 @@ import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse;
import org.apache.kafka.common.requests.DeleteGroupsResponse;
import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.requests.DeleteTopicsRequest;
+import org.apache.kafka.common.requests.DeleteTopicsResponse;
import org.apache.kafka.common.requests.DescribeAclsResponse;
import org.apache.kafka.common.requests.DescribeConfigsResponse;
import org.apache.kafka.common.requests.DescribeGroupsResponse;
@@ -379,6 +382,32 @@ public class KafkaAdminClientTest {
}
@Test
+ public void testDeleteTopics() throws Exception {
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ env.kafkaClient().setNode(env.cluster().controller());
+
+ env.kafkaClient().prepareResponse(body -> body instanceof DeleteTopicsRequest,
+ new DeleteTopicsResponse(Collections.singletonMap("myTopic", Errors.NONE)));
+ KafkaFuture<Void> future = env.adminClient().deleteTopics(Collections.singletonList("myTopic"),
+ new DeleteTopicsOptions()).all();
+ future.get();
+
+ env.kafkaClient().prepareResponse(body -> body instanceof DeleteTopicsRequest,
+ new DeleteTopicsResponse(Collections.singletonMap("myTopic", Errors.TOPIC_DELETION_DISABLED)));
+ future = env.adminClient().deleteTopics(Collections.singletonList("myTopic"),
+ new DeleteTopicsOptions()).all();
+ TestUtils.assertFutureError(future, TopicDeletionDisabledException.class);
+
+ env.kafkaClient().prepareResponse(body -> body instanceof DeleteTopicsRequest,
+ new DeleteTopicsResponse(Collections.singletonMap("myTopic", Errors.UNKNOWN_TOPIC_OR_PARTITION)));
+ future = env.adminClient().deleteTopics(Collections.singletonList("myTopic"),
+ new DeleteTopicsOptions()).all();
+ TestUtils.assertFutureError(future, UnknownTopicOrPartitionException.class);
+ }
+ }
+
+ @Test
public void testInvalidTopicNames() throws Exception {
try (AdminClientUnitTestEnv env = mockClientEnv()) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 0273e3d..096ff38 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -1539,6 +1539,12 @@ class KafkaApis(val requestChannel: RequestChannel,
(topic, Errors.NOT_CONTROLLER)
}.toMap
sendResponseCallback(results)
+ } else if (!config.deleteTopicEnable) {
+ val error = if (request.context.apiVersion < 3) Errors.INVALID_REQUEST else Errors.TOPIC_DELETION_DISABLED
+ val results = deleteTopicRequest.topics.asScala.map { topic =>
+ (topic, error)
+ }.toMap
+ sendResponseCallback(results)
} else {
// If no authorized topics return immediately
if (authorizedForDeleteTopics.isEmpty)
diff --git a/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala b/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala
new file mode 100644
index 0000000..20e30c0
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala
@@ -0,0 +1,59 @@
+/**
+ * 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.{ApiKeys, Errors}
+import org.apache.kafka.common.requests.{DeleteTopicsRequest, DeleteTopicsResponse}
+import org.junit.Assert._
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class DeleteTopicsRequestWithDeletionDisabledTest extends BaseRequestTest {
+
+ override def numBrokers: Int = 1
+
+ override def generateConfigs = {
+ val props = TestUtils.createBrokerConfigs(numBrokers, zkConnect,
+ enableControlledShutdown = false, enableDeleteTopic = false,
+ interBrokerSecurityProtocol = Some(securityProtocol),
+ trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount)
+ props.foreach(propertyOverrides)
+ props.map(KafkaConfig.fromProps)
+ }
+
+ @Test
+ def testDeleteRecordsRequest() {
+ val topic = "topic-1"
+ val request = new DeleteTopicsRequest.Builder(Set(topic).asJava, 1000).build()
+ val response = sendDeleteTopicsRequest(request)
+ assertEquals(Errors.TOPIC_DELETION_DISABLED, response.errors.get(topic))
+
+ val v2request = new DeleteTopicsRequest.Builder(Set(topic).asJava, 1000).build(2)
+ val v2response = sendDeleteTopicsRequest(v2request)
+ assertEquals(Errors.INVALID_REQUEST, v2response.errors.get(topic))
+ }
+
+ private def sendDeleteTopicsRequest(request: DeleteTopicsRequest, socketServer: SocketServer = controllerSocketServer): DeleteTopicsResponse = {
+ val response = connectAndSend(request, ApiKeys.DELETE_TOPICS, socketServer)
+ DeleteTopicsResponse.parse(response, request.version)
+ }
+
+}