KREST 9773 Handle InvalidTopicException correctly

This is kafka exception is raised due to bad user input, hence should be returned as a 400, instead of a 500(server error).
confluentinc · Mar 30, 2023 · 1ec7602 · 1ec7602
1 parent bdf15ab
commit 1ec7602
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/core/src/main/java/io/confluent/rest/exceptions/KafkaExceptionMapper.java b/core/src/main/java/io/confluent/rest/exceptions/KafkaExceptionMapper.java
@@ -27,6 +27,7 @@
 import org.apache.kafka.common.errors.InvalidPartitionsException;
 import org.apache.kafka.common.errors.InvalidReplicationFactorException;
 import org.apache.kafka.common.errors.InvalidRequestException;
+import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.PolicyViolationException;
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
@@ -126,6 +127,8 @@ private Response handleException(final Throwable exception) {
       }
       return getResponse(exception, Status.INTERNAL_SERVER_ERROR,
           KAFKA_RETRIABLE_ERROR_ERROR_CODE);
+    } else if (exception instanceof InvalidTopicException) {
+      return getResponse(exception, Status.BAD_REQUEST, KAFKA_BAD_REQUEST_ERROR_CODE);
     } else if (exception instanceof KafkaException) {
       log.error("Kafka exception", exception);
       return getResponse(exception, Status.INTERNAL_SERVER_ERROR,

diff --git a/core/src/test/java/io/confluent/rest/KafkaExceptionMapperTest.java b/core/src/test/java/io/confluent/rest/KafkaExceptionMapperTest.java
@@ -30,6 +30,7 @@
 import org.apache.kafka.common.errors.InvalidPartitionsException;
 import org.apache.kafka.common.errors.InvalidReplicationFactorException;
 import org.apache.kafka.common.errors.InvalidRequestException;
+import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.NotCoordinatorException;
 import org.apache.kafka.common.errors.NotEnoughReplicasException;
 import org.apache.kafka.common.errors.PolicyViolationException;
@@ -110,7 +111,7 @@ public void testKafkaExceptions() {
         KAFKA_BAD_REQUEST_ERROR_CODE);
     verifyMapperResponse(new InvalidRequestException("some message"), Status.BAD_REQUEST,
         KAFKA_BAD_REQUEST_ERROR_CODE);
-    verifyMapperResponse(new UnknownServerException("some message"),Status.BAD_REQUEST,
+    verifyMapperResponse(new UnknownServerException("some message"), Status.BAD_REQUEST,
         KAFKA_BAD_REQUEST_ERROR_CODE);
     verifyMapperResponse(new UnknownTopicOrPartitionException("some message"), Status.NOT_FOUND,
         KAFKA_UNKNOWN_TOPIC_PARTITION_CODE);
@@ -120,11 +121,14 @@ public void testKafkaExceptions() {
         KAFKA_BAD_REQUEST_ERROR_CODE);
     verifyMapperResponse(new InvalidConfigurationException("some message"), Status.BAD_REQUEST,
         KAFKA_BAD_REQUEST_ERROR_CODE);
+    verifyMapperResponse(new InvalidTopicException("some message"), Status.BAD_REQUEST,
+        KAFKA_BAD_REQUEST_ERROR_CODE);
 
     //test couple of retriable exceptions
     verifyMapperResponse(new NotCoordinatorException("some message"), Status.INTERNAL_SERVER_ERROR,
         KAFKA_RETRIABLE_ERROR_ERROR_CODE);
-    verifyMapperResponse(new NotEnoughReplicasException("some message"), Status.INTERNAL_SERVER_ERROR,
+    verifyMapperResponse(new NotEnoughReplicasException("some message"),
+        Status.INTERNAL_SERVER_ERROR,
         KAFKA_RETRIABLE_ERROR_ERROR_CODE);
     //Including the special case of a topic not being present (eg because it's not been defined yet)
     //not returning a 500 error