[strimzi#434] feat: added create topic endpoint

Signed-off-by: ilkerkocatepe <[email protected]>

Author: ilkerkocatepe

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

@@ -9,13 +9,15 @@
 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.Properties;
@@ -84,6 +86,29 @@ public CompletionStage<Set<String>> listTopics() {
         return promise;
     }
 
+    /**
+     * Creates a topic with given name
+     *
+     * @param topicName topic name to create
+     * @return a CompletionStage Void
+     */
+    public CompletionStage<Void> createTopic(String topicName) {
+        log.trace("Create topic thread {}", Thread.currentThread());
+        log.info("Create topic {}", topicName);
+        CompletableFuture<Void> promise = new CompletableFuture<>();
+        this.adminClient.createTopics(Collections.singletonList(new NewTopic(topicName, 2, (short) 1)))
+                .all()
+                .whenComplete((topic, exception) -> {
+                    log.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

@@ -81,6 +81,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 +177,32 @@ public void doGetTopic(RoutingContext routingContext) {
                 });
     }
 
+    /**
+     * Create a topic described in the HTTP request
+     *
+     * @param routingContext the routing context
+     */
+    public void doCreateTopic(RoutingContext routingContext) {
+        String topicName = routingContext.pathParam("topicname");
+
+        this.kafkaBridgeAdmin.createTopic(topicName)
+                .whenComplete(((topic, exception) -> {
+                    log.trace("Create topic handler thread {}", Thread.currentThread());
+                    if (exception == null) {
+                        ArrayNode root = JsonUtils.createArrayNode();
+                        HttpUtils.sendResponse(routingContext, HttpResponseStatus.OK.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 */

src/main/resources/openapi.json

@@ -753,6 +753,56 @@
                 }
             ]
         },
+        "/create-topic/{topicname}": {
+            "post": {
+                "tags": [
+                    "Topics"
+                ],
+                "description": "Creates a topic with given topic name.",
+                "operationId": "createTopic",
+                "responses": {
+                    "200": {
+                        "description": "Topic metadata",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/TopicMetadata"
+                                }
+                            }
+                        }
+                    },
+                    "404": {
+                        "description": "The specified topic was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified topic was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "topicname",
+                    "in": "path",
+                    "description": "Name of the topic to send records to or retrieve metadata from.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
+        },
         "/consumers/{groupid}/instances/{name}/records": {
             "get": {
                 "tags": [

src/test/java/io/strimzi/kafka/bridge/http/AdminClientIT.java

@@ -215,4 +215,19 @@ void setupTopic(VertxTestContext context, String topic, int partitions, int coun
                 });
         producer.get(TEST_TIMEOUT, TimeUnit.SECONDS);
     }
+
+    @Test
+    void createTopicTest(VertxTestContext context) {
+        baseService()
+                .postRequest("/create-topic/" + topic)
+                .as(BodyCodec.jsonArray())
+                .send(ar -> {
+                    context.verify(() -> {
+                        assertThat(ar.succeeded(), is(true));
+                        HttpResponse<JsonArray> response = ar.result();
+                        assertThat(response.statusCode(), is(HttpResponseStatus.OK.code()));
+                    });
+                    context.completeNow();
+                });
+    }
 }