[#434] feat: added create topic endpoint

CHANGELOG.md

@@ -2,7 +2,8 @@
 
 ## 0.31.0
 
-* Dependency updates (Vert.x 4.5.10)
+* Dependency updates (Kafka 3.9.0, Vert.x 4.5.10)
+* Added support for creating a new topic via endpoint.
 
 ## 0.30.0
 

documentation/book/api/.openapi-generator/openapi.json-generate-apidoc.sha256

@@ -1 +1 @@
-e4a70d99cdac5e8dddb45ce30c39817fa9402463074478e0fd3a05d6a7f849d5
+cf88d8909114896517ba4027596382126bca594d55dd9a924330232407e603b2

documentation/book/api/index.adoc

@@ -2289,6 +2289,87 @@ endif::internal-generation[]
 === Topics
 
 
+[.createTopic]
+==== createTopic
+
+`POST /admin/topics`
+
+
+
+===== Description
+
+Creates a topic with given name, partitions count, and replication factor.
+
+
+// markup not found, no include::{specDir}admin/topics/POST/spec.adoc[opts=optional]
+
+
+
+===== Parameters
+
+
+
+[cols="2,3,1,1,1"]
+.Body Parameter
+|===
+|Name| Description| Required| Default| Pattern
+
+| NewTopic
+| Creates a topic with given name, partitions count, and replication factor. <<NewTopic>>
+| X
+| 
+| 
+
+|===
+
+
+
+
+
+===== Return Type
+
+
+
+
+-
+
+
+===== Responses
+
+.HTTP Response Codes
+[cols="2,3,1"]
+|===
+| Code | Message | Datatype
+
+
+| 201
+| Created
+|  <<>>
+
+|===
+
+===== Samples
+
+
+include::{snippetDir}admin/topics/POST/http-request.adoc[opts=optional]
+
+
+// markup not found, no include::{snippetDir}admin/topics/POST/http-response.adoc[opts=optional]
+
+
+
+// file not found, no * wiremock data link :admin/topics/POST/POST.json[]
+
+
+ifdef::internal-generation[]
+===== Implementation
+
+// markup not found, no include::{specDir}admin/topics/POST/implementation.adoc[opts=optional]
+
+
+endif::internal-generation[]
+
+
 [.getOffsets]
 ==== getOffsets
 
@@ -3209,6 +3290,42 @@ Information about Kafka Bridge instance.
 
 
 
+[#NewTopic]
+=== _NewTopic_ NewTopic
+
+
+
+
+[.fields-NewTopic]
+[cols="2,1,1,2,4,1"]
+|===
+| Field Name| Required| Nullable | Type| Description | Format
+
+| topic_name
+| X
+| 
+|   String  
+| Name of the topic to create.
+|     
+
+| partitions_count
+| 
+| X
+|   Integer  
+| Number of partitions for the topic.
+|     
+
+| replication_factor
+| 
+| X
+|   Integer  
+| Number of replicas for each partition.
+|     
+
+|===
+
+
+
 [#OffsetCommitSeek]
 === _OffsetCommitSeek_ OffsetCommitSeek
 

documentation/book/api/snippet/admin/topics/POST/http-request.adoc

@@ -0,0 +1,11 @@
+==== Example HTTP request
+
+===== Request body
+[source,json]
+----
+{
+  "topic_name" : "my-topic",
+  "partitions_count" : 1,
+  "replication_factor" : 2,
+}
+----

pom.xml

@@ -102,8 +102,8 @@
 		<log4j.version>2.17.2</log4j.version>
 		<vertx.version>4.5.10</vertx.version>
 		<vertx-testing.version>4.5.10</vertx-testing.version>
-		<netty.version>4.1.111.Final</netty.version>
-		<kafka.version>3.8.0</kafka.version>
+		<netty.version>4.1.115.Final</netty.version>
+		<kafka.version>3.9.0</kafka.version>
 		<kafka-kubernetes-config-provider.version>1.2.0</kafka-kubernetes-config-provider.version>
 		<kafka-env-var-config-provider.version>1.1.0</kafka-env-var-config-provider.version>
 		<maven.checkstyle.version>3.3.0</maven.checkstyle.version>

src/main/java/io/strimzi/kafka/bridge/KafkaBridgeAdmin.java

@@ -9,15 +9,18 @@
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -84,6 +87,31 @@ public CompletionStage<Set<String>> listTopics() {
         return promise;
     }
 
+    /**
+     * Creates a topic with given name and number of partitions (optional) and replication factor (optional).
+     *
+     * @param topicName topic name to create
+     * @param partitions number of partitions
+     * @param replicationFactor replication factor
+     * @return a CompletionStage Void
+     */
+    public CompletionStage<Void> createTopic(String topicName, Optional<Integer> partitions, Optional<Short> replicationFactor) {
+        LOGGER.trace("Create topic thread {}", Thread.currentThread());
+        LOGGER.info("Create topic {}, partitions {}, replicationFactor {}", topicName, partitions, replicationFactor);
+        CompletableFuture<Void> promise = new CompletableFuture<>();
+        this.adminClient.createTopics(Collections.singletonList(new NewTopic(topicName, partitions, replicationFactor)))
+                .all()
+                .whenComplete((topic, exception) -> {
+                    LOGGER.trace("Create topic callback thread {}", Thread.currentThread());
+                    if (exception == null) {
+                        promise.complete(topic);
+                    } else {
+                        promise.completeExceptionally(exception);
+                    }
+                });
+        return promise;
+    }
+
     /**
      * Returns the description of the specified topics.
      *

src/main/java/io/strimzi/kafka/bridge/http/HttpAdminBridgeEndpoint.java

@@ -15,6 +15,7 @@
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.http.converter.JsonUtils;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
+import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.RoutingContext;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
@@ -32,6 +33,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -81,6 +83,10 @@ public void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> ha
                 doGetTopic(routingContext);
                 break;
 
+            case CREATE_TOPIC:
+                doCreateTopic(routingContext);
+                break;
+
             case LIST_PARTITIONS:
                 doListPartitions(routingContext);
                 break;
@@ -173,6 +179,47 @@ public void doGetTopic(RoutingContext routingContext) {
                 });
     }
 
+    /**
+     * Create a topic with described name, partitions count, and replication factor in the body of the HTTP request
+     *
+     * @param routingContext the routing context
+     */
+    public void doCreateTopic(RoutingContext routingContext) {
+        JsonObject jsonBody = routingContext.body().asJsonObject();
+
+        if (jsonBody.isEmpty()) {
+            HttpBridgeError error = new HttpBridgeError(
+                    HttpResponseStatus.UNPROCESSABLE_ENTITY.code(),
+                    "Request body must be a JSON object"
+            );
+            HttpUtils.sendResponse(routingContext, HttpResponseStatus.UNPROCESSABLE_ENTITY.code(),
+                    BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
+            return;
+        }
+
+        String topicName = jsonBody.getString("topic_name");
+        Optional<Integer> partitionsCount = Optional.ofNullable(jsonBody.getInteger("partitions_count"));
+        Optional<Short> replicationFactor = Optional.ofNullable(jsonBody.getInteger("replication_factor"))
+                .map(Integer::shortValue);
+
+        this.kafkaBridgeAdmin.createTopic(topicName, partitionsCount, replicationFactor)
+                .whenComplete(((topic, exception) -> {
+                    LOGGER.trace("Create topic handler thread {}", Thread.currentThread());
+                    if (exception == null) {
+                        JsonNode root = JsonUtils.createObjectNode();
+                        HttpUtils.sendResponse(routingContext, HttpResponseStatus.CREATED.code(),
+                                BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(root));
+                    } else {
+                        HttpBridgeError error = new HttpBridgeError(
+                                HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                                exception.getMessage()
+                        );
+                        HttpUtils.sendResponse(routingContext, HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                                BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
+                    }
+                }));
+    }
+
     /**
      * Get partitions information related to the topic in the HTTP request
      *

src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java

@@ -157,6 +157,7 @@ public void start(Promise<Void> startPromise) {
                 routerBuilder.operation(this.SEEK_TO_END.getOperationId().toString()).handler(this.SEEK_TO_END);
                 routerBuilder.operation(this.LIST_TOPICS.getOperationId().toString()).handler(this.LIST_TOPICS);
                 routerBuilder.operation(this.GET_TOPIC.getOperationId().toString()).handler(this.GET_TOPIC);
+                routerBuilder.operation(this.CREATE_TOPIC.getOperationId().toString()).handler(this.CREATE_TOPIC);
                 routerBuilder.operation(this.LIST_PARTITIONS.getOperationId().toString()).handler(this.LIST_PARTITIONS);
                 routerBuilder.operation(this.GET_PARTITION.getOperationId().toString()).handler(this.GET_PARTITION);
                 routerBuilder.operation(this.GET_OFFSETS.getOperationId().toString()).handler(this.GET_OFFSETS);
@@ -381,6 +382,11 @@ private void getTopic(RoutingContext routingContext) {
         processAdminClient(routingContext);
     }
 
+    private void createTopic(RoutingContext routingContext) {
+        this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.CREATE_TOPIC);
+        processAdminClient(routingContext);
+    }
+
     private void listPartitions(RoutingContext routingContext) {
         this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.LIST_PARTITIONS);
         processAdminClient(routingContext);
@@ -726,6 +732,14 @@ public void process(RoutingContext routingContext) {
         }
     };
 
+    final HttpOpenApiOperation CREATE_TOPIC = new HttpOpenApiOperation(HttpOpenApiOperations.CREATE_TOPIC) {
+
+        @Override
+        public void process(RoutingContext routingContext) {
+            createTopic(routingContext);
+        }
+    };
+
     final HttpOpenApiOperation LIST_PARTITIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_PARTITIONS) {
 
         @Override

src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java

@@ -28,6 +28,8 @@ public enum HttpOpenApiOperations {
     LIST_TOPICS("listTopics"),
     /** get information for a specific topic */
     GET_TOPIC("getTopic"),
+    /** creates a topic with specified name */
+    CREATE_TOPIC("createTopic"),
     /** list partitions for a specific topic */
     LIST_PARTITIONS("listPartitions"),
     /** get partition information for a specific topic */