Skip to content

Commit fa55d1b

Browse files
ilkerkocatepeppatiernoscholzjdependabot[bot]
authored
[#434] feat: added create topic endpoint (#929)
* [#434] feat: added create topic endpoint Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - logger edit Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - logger edit Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - openapiv2.json edits and tests Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - path changed Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - javadoc fix Signed-off-by: ilkerkocatepe <[email protected]> * [#434] Update generated documentation Signed-off-by: ilkerkocatepe <[email protected]> * Bumped Snakeyaml to 2.2 (#930) Signed-off-by: Paolo Patierno <[email protected]> * Update Kubernetes Config Provider to 1.2.0 (#931) Signed-off-by: Jakub Scholz <[email protected]> * Excluded android related dependency (#933) Signed-off-by: Paolo Patierno <[email protected]> * [#434] fix: after review - partitions and replication_factor parameters set optional Signed-off-by: ilkerkocatepe <[email protected]> * [#434] Update generated documentation Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: after review - partitions and replication_factor parameters set optional Signed-off-by: ilkerkocatepe <[email protected]> * [#434] Update generated documentation Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: after review fix Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: checkstyle edits Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: checkstyle edits Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: method description detailed Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: removed unnecessary response code Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: removed unnecessary response code Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: docs edited after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: edits for api standards Signed-off-by: ilkerkocatepe <[email protected]> * Bump Kafka 3.8.1 (#935) Signed-off-by: Paolo Patierno <[email protected]> * [#434] fix: edits after review Signed-off-by: ilkerkocatepe <[email protected]> * Bumped Kafka 3.9.0 (#936) Signed-off-by: Paolo Patierno <[email protected]> * Bump io.netty:netty-common from 4.1.111.Final to 4.1.115.Final (#937) Bumps [io.netty:netty-common](https://github.com/netty/netty) from 4.1.111.Final to 4.1.115.Final. - [Commits](netty/netty@netty-4.1.111.Final...netty-4.1.115.Final) --- updated-dependencies: - dependency-name: io.netty:netty-common dependency-type: direct:production ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Bumped Vert.x 4.5.11 (#938) Signed-off-by: Paolo Patierno <[email protected]> * [#434] docs: edits after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: edits after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: edits after review & changelog Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - logger edit Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - logger edit Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - openapiv2.json edits and tests Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - path changed Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: added create topic endpoint - javadoc fix Signed-off-by: ilkerkocatepe <[email protected]> * [#434] Update generated documentation Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: after review - partitions and replication_factor parameters set optional Signed-off-by: ilkerkocatepe <[email protected]> * [#434] Update generated documentation Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: after review - partitions and replication_factor parameters set optional Signed-off-by: ilkerkocatepe <[email protected]> * [#434] Update generated documentation Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: after review fix Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: checkstyle edits Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: checkstyle edits Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: method description detailed Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: removed unnecessary response code Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: removed unnecessary response code Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: docs edited after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] feat: edits for api standards Signed-off-by: ilkerkocatepe <[email protected]> * [#434] fix: edits after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: edits after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: edits after review Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: edits after review & changelog Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: changelog Signed-off-by: ilkerkocatepe <[email protected]> * Revert "Bumped Vert.x 4.5.11 (#938)" This reverts commit ea03491 Signed-off-by: ilkerkocatepe <[email protected]> * [#434] docs: changelog Signed-off-by: ilkerkocatepe <[email protected]> --------- Signed-off-by: ilkerkocatepe <[email protected]> Signed-off-by: Paolo Patierno <[email protected]> Signed-off-by: Jakub Scholz <[email protected]> Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: Paolo Patierno <[email protected]> Co-authored-by: Jakub Scholz <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
1 parent ea03491 commit fa55d1b

File tree

13 files changed

+392
-5
lines changed

13 files changed

+392
-5
lines changed

‎CHANGELOG.md

+2-1
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@
22

33
## 0.31.0
44

5-
* Dependency updates (Kafka 3.9.0, Vert.x 4.5.11, Netty 4.1.115.Final)
5+
* Dependency updates (Kafka 3.9.0, Vert.x 4.5.10)
6+
* Added support for creating a new topic via endpoint.
67

78
## 0.30.0
89

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
e4a70d99cdac5e8dddb45ce30c39817fa9402463074478e0fd3a05d6a7f849d5
1+
cf88d8909114896517ba4027596382126bca594d55dd9a924330232407e603b2

‎documentation/book/api/index.adoc

+117
Original file line numberDiff line numberDiff line change
@@ -2289,6 +2289,87 @@ endif::internal-generation[]
22892289
=== Topics
22902290

22912291

2292+
[.createTopic]
2293+
==== createTopic
2294+
2295+
`POST /admin/topics`
2296+
2297+
2298+
2299+
===== Description
2300+
2301+
Creates a topic with given name, partitions count, and replication factor.
2302+
2303+
2304+
// markup not found, no include::{specDir}admin/topics/POST/spec.adoc[opts=optional]
2305+
2306+
2307+
2308+
===== Parameters
2309+
2310+
2311+
2312+
[cols="2,3,1,1,1"]
2313+
.Body Parameter
2314+
|===
2315+
|Name| Description| Required| Default| Pattern
2316+
2317+
| NewTopic
2318+
| Creates a topic with given name, partitions count, and replication factor. <<NewTopic>>
2319+
| X
2320+
|
2321+
|
2322+
2323+
|===
2324+
2325+
2326+
2327+
2328+
2329+
===== Return Type
2330+
2331+
2332+
2333+
2334+
-
2335+
2336+
2337+
===== Responses
2338+
2339+
.HTTP Response Codes
2340+
[cols="2,3,1"]
2341+
|===
2342+
| Code | Message | Datatype
2343+
2344+
2345+
| 201
2346+
| Created
2347+
| <<>>
2348+
2349+
|===
2350+
2351+
===== Samples
2352+
2353+
2354+
include::{snippetDir}admin/topics/POST/http-request.adoc[opts=optional]
2355+
2356+
2357+
// markup not found, no include::{snippetDir}admin/topics/POST/http-response.adoc[opts=optional]
2358+
2359+
2360+
2361+
// file not found, no * wiremock data link :admin/topics/POST/POST.json[]
2362+
2363+
2364+
ifdef::internal-generation[]
2365+
===== Implementation
2366+
2367+
// markup not found, no include::{specDir}admin/topics/POST/implementation.adoc[opts=optional]
2368+
2369+
2370+
endif::internal-generation[]
2371+
2372+
22922373
[.getOffsets]
22932374
==== getOffsets
22942375

@@ -3209,6 +3290,42 @@ Information about Kafka Bridge instance.
32093290

32103291

32113292

3293+
[#NewTopic]
3294+
=== _NewTopic_ NewTopic
3295+
3296+
3297+
3298+
3299+
[.fields-NewTopic]
3300+
[cols="2,1,1,2,4,1"]
3301+
|===
3302+
| Field Name| Required| Nullable | Type| Description | Format
3303+
3304+
| topic_name
3305+
| X
3306+
|
3307+
| String
3308+
| Name of the topic to create.
3309+
|
3310+
3311+
| partitions_count
3312+
|
3313+
| X
3314+
| Integer
3315+
| Number of partitions for the topic.
3316+
|
3317+
3318+
| replication_factor
3319+
|
3320+
| X
3321+
| Integer
3322+
| Number of replicas for each partition.
3323+
|
3324+
3325+
|===
3326+
3327+
3328+
32123329
[#OffsetCommitSeek]
32133330
=== _OffsetCommitSeek_ OffsetCommitSeek
32143331

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

+11
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
==== Example HTTP request
2+
3+
===== Request body
4+
[source,json]
5+
----
6+
{
7+
"topic_name" : "my-topic",
8+
"partitions_count" : 1,
9+
"replication_factor" : 2,
10+
}
11+
----

‎pom.xml

+2-2
Original file line numberDiff line numberDiff line change
@@ -100,8 +100,8 @@
100100
<maven.compiler.source>11</maven.compiler.source>
101101
<maven.compiler.target>11</maven.compiler.target>
102102
<log4j.version>2.17.2</log4j.version>
103-
<vertx.version>4.5.11</vertx.version>
104-
<vertx-testing.version>4.5.11</vertx-testing.version>
103+
<vertx.version>4.5.10</vertx.version>
104+
<vertx-testing.version>4.5.10</vertx-testing.version>
105105
<netty.version>4.1.115.Final</netty.version>
106106
<kafka.version>3.9.0</kafka.version>
107107
<kafka-kubernetes-config-provider.version>1.2.0</kafka-kubernetes-config-provider.version>

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

+28
Original file line numberDiff line numberDiff line change
@@ -9,15 +9,18 @@
99
import org.apache.kafka.clients.admin.AdminClient;
1010
import org.apache.kafka.clients.admin.Config;
1111
import org.apache.kafka.clients.admin.ListOffsetsResult;
12+
import org.apache.kafka.clients.admin.NewTopic;
1213
import org.apache.kafka.clients.admin.OffsetSpec;
1314
import org.apache.kafka.clients.admin.TopicDescription;
1415
import org.apache.kafka.common.TopicPartition;
1516
import org.apache.kafka.common.config.ConfigResource;
1617
import org.apache.logging.log4j.LogManager;
1718
import org.apache.logging.log4j.Logger;
1819

20+
import java.util.Collections;
1921
import java.util.List;
2022
import java.util.Map;
23+
import java.util.Optional;
2124
import java.util.Properties;
2225
import java.util.Set;
2326
import java.util.concurrent.CompletableFuture;
@@ -84,6 +87,31 @@ public CompletionStage<Set<String>> listTopics() {
8487
return promise;
8588
}
8689

90+
/**
91+
* Creates a topic with given name and number of partitions (optional) and replication factor (optional).
92+
*
93+
* @param topicName topic name to create
94+
* @param partitions number of partitions
95+
* @param replicationFactor replication factor
96+
* @return a CompletionStage Void
97+
*/
98+
public CompletionStage<Void> createTopic(String topicName, Optional<Integer> partitions, Optional<Short> replicationFactor) {
99+
LOGGER.trace("Create topic thread {}", Thread.currentThread());
100+
LOGGER.info("Create topic {}, partitions {}, replicationFactor {}", topicName, partitions, replicationFactor);
101+
CompletableFuture<Void> promise = new CompletableFuture<>();
102+
this.adminClient.createTopics(Collections.singletonList(new NewTopic(topicName, partitions, replicationFactor)))
103+
.all()
104+
.whenComplete((topic, exception) -> {
105+
LOGGER.trace("Create topic callback thread {}", Thread.currentThread());
106+
if (exception == null) {
107+
promise.complete(topic);
108+
} else {
109+
promise.completeExceptionally(exception);
110+
}
111+
});
112+
return promise;
113+
}
114+
87115
/**
88116
* Returns the description of the specified topics.
89117
*

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

+47
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import io.strimzi.kafka.bridge.config.BridgeConfig;
1616
import io.strimzi.kafka.bridge.http.converter.JsonUtils;
1717
import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
18+
import io.vertx.core.json.JsonObject;
1819
import io.vertx.ext.web.RoutingContext;
1920
import org.apache.kafka.clients.admin.Config;
2021
import org.apache.kafka.clients.admin.ConfigEntry;
@@ -32,6 +33,7 @@
3233
import java.util.HashSet;
3334
import java.util.List;
3435
import java.util.Map;
36+
import java.util.Optional;
3537
import java.util.Set;
3638
import java.util.concurrent.CompletableFuture;
3739
import java.util.concurrent.CompletionStage;
@@ -81,6 +83,10 @@ public void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> ha
8183
doGetTopic(routingContext);
8284
break;
8385

86+
case CREATE_TOPIC:
87+
doCreateTopic(routingContext);
88+
break;
89+
8490
case LIST_PARTITIONS:
8591
doListPartitions(routingContext);
8692
break;
@@ -173,6 +179,47 @@ public void doGetTopic(RoutingContext routingContext) {
173179
});
174180
}
175181

182+
/**
183+
* Create a topic with described name, partitions count, and replication factor in the body of the HTTP request
184+
*
185+
* @param routingContext the routing context
186+
*/
187+
public void doCreateTopic(RoutingContext routingContext) {
188+
JsonObject jsonBody = routingContext.body().asJsonObject();
189+
190+
if (jsonBody.isEmpty()) {
191+
HttpBridgeError error = new HttpBridgeError(
192+
HttpResponseStatus.UNPROCESSABLE_ENTITY.code(),
193+
"Request body must be a JSON object"
194+
);
195+
HttpUtils.sendResponse(routingContext, HttpResponseStatus.UNPROCESSABLE_ENTITY.code(),
196+
BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
197+
return;
198+
}
199+
200+
String topicName = jsonBody.getString("topic_name");
201+
Optional<Integer> partitionsCount = Optional.ofNullable(jsonBody.getInteger("partitions_count"));
202+
Optional<Short> replicationFactor = Optional.ofNullable(jsonBody.getInteger("replication_factor"))
203+
.map(Integer::shortValue);
204+
205+
this.kafkaBridgeAdmin.createTopic(topicName, partitionsCount, replicationFactor)
206+
.whenComplete(((topic, exception) -> {
207+
LOGGER.trace("Create topic handler thread {}", Thread.currentThread());
208+
if (exception == null) {
209+
JsonNode root = JsonUtils.createObjectNode();
210+
HttpUtils.sendResponse(routingContext, HttpResponseStatus.CREATED.code(),
211+
BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(root));
212+
} else {
213+
HttpBridgeError error = new HttpBridgeError(
214+
HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
215+
exception.getMessage()
216+
);
217+
HttpUtils.sendResponse(routingContext, HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
218+
BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
219+
}
220+
}));
221+
}
222+
176223
/**
177224
* Get partitions information related to the topic in the HTTP request
178225
*

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

+14
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,7 @@ public void start(Promise<Void> startPromise) {
157157
routerBuilder.operation(this.SEEK_TO_END.getOperationId().toString()).handler(this.SEEK_TO_END);
158158
routerBuilder.operation(this.LIST_TOPICS.getOperationId().toString()).handler(this.LIST_TOPICS);
159159
routerBuilder.operation(this.GET_TOPIC.getOperationId().toString()).handler(this.GET_TOPIC);
160+
routerBuilder.operation(this.CREATE_TOPIC.getOperationId().toString()).handler(this.CREATE_TOPIC);
160161
routerBuilder.operation(this.LIST_PARTITIONS.getOperationId().toString()).handler(this.LIST_PARTITIONS);
161162
routerBuilder.operation(this.GET_PARTITION.getOperationId().toString()).handler(this.GET_PARTITION);
162163
routerBuilder.operation(this.GET_OFFSETS.getOperationId().toString()).handler(this.GET_OFFSETS);
@@ -381,6 +382,11 @@ private void getTopic(RoutingContext routingContext) {
381382
processAdminClient(routingContext);
382383
}
383384

385+
private void createTopic(RoutingContext routingContext) {
386+
this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.CREATE_TOPIC);
387+
processAdminClient(routingContext);
388+
}
389+
384390
private void listPartitions(RoutingContext routingContext) {
385391
this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.LIST_PARTITIONS);
386392
processAdminClient(routingContext);
@@ -726,6 +732,14 @@ public void process(RoutingContext routingContext) {
726732
}
727733
};
728734

735+
final HttpOpenApiOperation CREATE_TOPIC = new HttpOpenApiOperation(HttpOpenApiOperations.CREATE_TOPIC) {
736+
737+
@Override
738+
public void process(RoutingContext routingContext) {
739+
createTopic(routingContext);
740+
}
741+
};
742+
729743
final HttpOpenApiOperation LIST_PARTITIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_PARTITIONS) {
730744

731745
@Override

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

+2
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@ public enum HttpOpenApiOperations {
2828
LIST_TOPICS("listTopics"),
2929
/** get information for a specific topic */
3030
GET_TOPIC("getTopic"),
31+
/** creates a topic with specified name */
32+
CREATE_TOPIC("createTopic"),
3133
/** list partitions for a specific topic */
3234
LIST_PARTITIONS("listPartitions"),
3335
/** get partition information for a specific topic */

0 commit comments

Comments
 (0)