Repository: strimzi/strimzi-kafka-bridge
Pull request #504 (2021-03-22 12:09:37)

Update vertx major version
Signed-off-by: Stanislav Knot <sknot@redhat.com>

Upgrade Vertx to version 4.0.3


Patch:
diff --git a/Dockerfile b/Dockerfile
index 9a90129..2c5cd49 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,5 +1,6 @@
 FROM registry.access.redhat.com/ubi8/ubi-minimal:latest
 ARG JAVA_VERSION=11
+ARG TARGETPLATFORM
 
 USER root
 
diff --git a/HACKING.md b/HACKING.md
index 873e055..1d3a183 100644
--- a/HACKING.md
+++ b/HACKING.md
@@ -75,3 +75,16 @@ Attach the ZIP and TAR.GZ archives to the release.
 8. On the `main` git branch:
   * Update the versions to the next SNAPSHOT version using the `next_version` `make` target. 
   For example to update the next version to `0.6.0-SNAPSHOT` run: `make NEXT_VERSION=0.6.0-SNAPSHOT next_version`.
+
+## Building container images for other platforms with Docker `buildx`
+
+Docker supports building images for different platforms using the `docker buildx` command. If you want to use it to
+build Strimzi images, you can just set the environment variable `DOCKER_BUILDX` to `buildx`, set the environment
+variable `DOCKER_BUILD_ARGS` to pass additional build options such as the platform and run the build. For example
+following can be used to build Strimzi images for Linux on PowerPC/ppc64le architecture:
+
+```
+export DOCKER_BUILDX=buildx
+export DOCKER_BUILD_ARGS="--platform linux/ppc64le"
+make all
+```
diff --git a/Makefile.docker b/Makefile.docker
index ffa8d77..feb7184 100644
--- a/Makefile.docker
+++ b/Makefile.docker
@@ -16,7 +16,7 @@ RELEASE_VERSION    ?= $(shell cat $(TOPDIR)/release.version)
 .PHONY: docker_build
 docker_build:
 	# Build Docker image ...
-	docker build $(DOCKER_BUILD_ARGS) --build-arg strimzi_kafka_bridge_version=$(RELEASE_VERSION) -t strimzi/$(PROJECT_NAME):latest $(DOCKERFILE_DIR)
+	docker $(DOCKER_BUILDX) build $(DOCKER_BUILD_ARGS) --build-arg strimzi_kafka_bridge_version=$(RELEASE_VERSION) -t strimzi/$(PROJECT_NAME):latest $(DOCKERFILE_DIR)
 #   The Dockerfiles all use FROM ...:latest, so it is necessary to tag images with latest (-t above)
 #   But because we generate Kafka images for different versions we also need to tag with something
 #   including the kafka version number. This BUILD_TAG is used by the docker_tag target.
@@ -31,4 +31,4 @@ docker_tag:
 .PHONY: docker_push
 docker_push: docker_tag
 	# Push the $(DOCKER_TAG)-tagged image to the registry
-	docker push $(DOCKER_REGISTRY)/$(DOCKER_ORG)/$(PROJECT_NAME):$(DOCKER_TAG)
\ No newline at end of file
+	docker push $(DOCKER_REGISTRY)/$(DOCKER_ORG)/$(PROJECT_NAME):$(DOCKER_TAG)
diff --git a/pom.xml b/pom.xml
index 276b4bd..b16d79c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -13,7 +13,7 @@
 		<maven.compiler.source>1.8</maven.compiler.source>
 		<maven.compiler.target>1.8</maven.compiler.target>
 		<log4j.version>2.13.3</log4j.version>
-		<vertx.version>3.9.4</vertx.version>
+		<vertx.version>4.0.3</vertx.version>
 		<debezium.version>1.2.3.Final</debezium.version>
 		<maven.checkstyle.version>3.1.0</maven.checkstyle.version>
 		<hamcrest.version>2.2</hamcrest.version>
diff --git a/src/main/java/io/strimzi/kafka/bridge/Application.java b/src/main/java/io/strimzi/kafka/bridge/Application.java
index ef92544..9f3934e 100644
--- a/src/main/java/io/strimzi/kafka/bridge/Application.java
+++ b/src/main/java/io/strimzi/kafka/bridge/Application.java
@@ -42,8 +42,10 @@ import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.EnumSet;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 /**
@@ -160,12 +162,15 @@ public class Application {
      * @return instance of the MicrometerMetricsOptions on Vert.x
      */
     private static MicrometerMetricsOptions metricsOptions() {
+        Set<String> set = new HashSet();
+        set.add(MetricsDomain.NAMED_POOLS.name());
+        set.add(MetricsDomain.VERTICLES.name());
         return new MicrometerMetricsOptions()
                 .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
                 // define the labels on the HTTP server related metrics
                 .setLabels(EnumSet.of(Label.REMOTE, Label.LOCAL, Label.HTTP_PATH, Label.HTTP_METHOD, Label.HTTP_CODE))
                 // disable metrics about pool and verticles
-                .setDisabledMetricsCategories(EnumSet.of(MetricsDomain.NAMED_POOLS, MetricsDomain.VERTICLES))
+                .setDisabledMetricsCategories(set)
                 .setJvmMetricsEnabled(true)
                 .setEnabled(true);
     }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
index b4a9fab..8cd6943 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -274,7 +274,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.CREATE_CONSUMER);
 
         // check for an empty body
-        JsonObject body = routingContext.getBody().length() != 0 ? routingContext.getBodyAsJson() : new JsonObject();
+        JsonObject body = routingContext.getBody() != null && routingContext.getBody().length() != 0 ? routingContext.getBodyAsJson() : new JsonObject();
         SinkBridgeEndpoint<byte[], byte[]> sink = null;
 
         try {
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
index 2ab080b..1d7bf08 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -483,6 +483,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         }
     }
 
+    @SuppressWarnings({"checkstyle:CyclomaticComplexity"})
     @Override
     public void handle(Endpoint<?> endpoint, Handler<?> handler) {
         RoutingContext routingContext = (RoutingContext) endpoint.get();
@@ -490,7 +491,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         JsonObject bodyAsJson = EMPTY_JSON;
         try {
             // check for an empty body
-            if (routingContext.getBody().length() != 0) {
+            if (routingContext.getBody() != null && routingContext.getBody().length() != 0) {
                 bodyAsJson = routingContext.getBodyAsJson();
             }
             log.debug("[{}] Request: body = {}", routingContext.get("request-id"), bodyAsJson);
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpBinaryMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpBinaryMessageConverter.java
index f4beca2..127eaf0 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpBinaryMessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpBinaryMessageConverter.java
@@ -44,7 +44,7 @@ public class HttpBinaryMessageConverter implements MessageConverter<byte[], byte
                     JsonObject jsonObject = (JsonObject) obj;
                     headers.add(new KafkaHeaderImpl(
                         jsonObject.getString("key"),
-                        Buffer.factory.buffer(
+                        Buffer.buffer(
                             DatatypeConverter.parseBase64Binary(jsonObject.getString("value")))));
                 }
             }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java
index fd16ddd..eccad46 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java
@@ -44,7 +44,7 @@ public class HttpJsonMessageConverter implements MessageConverter<byte[], byte[]
                     JsonObject jsonObject = (JsonObject) obj;
                     headers.add(new KafkaHeaderImpl(
                         jsonObject.getString("key"),
-                        Buffer.factory.buffer(
+                        Buffer.buffer(
                             DatatypeConverter.parseBase64Binary(jsonObject.getString("value")))));
                 }
             }
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index de707fc..7356bdf 100644
--- a/src/main/resources/openapi.json
+++ b/src/main/resources/openapi.json
@@ -271,7 +271,27 @@
                     }
                 },
                 "operationId": "listSubscriptions",
-                "description": "Retrieves a list of the topics to which the consumer is subscribed."
+                "description": "Retrieves a list of the topics to which the consumer is subscribed.",
+                "parameters": [
+                    {
+                        "name": "groupid",
+                        "in": "path",
+                        "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
+                    },
+                    {
+                        "name": "name",
+                        "in": "path",
+                        "description": "Name of the subscribed consumer.",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
+                    }
+                ]
             },
             "post": {
                 "tags": [
@@ -348,7 +368,27 @@
                             }
                         }
                     }
-                }
+                },
+                "parameters": [
+                    {
+                        "name": "groupid",
+                        "in": "path",
+                        "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
+                    },
+                    {
+                        "name": "name",
+                        "in": "path",
+                        "description": "Name of the consumer to subscribe to topics.",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
+                    }
+                ]
             },
             "delete": {
                 "tags": [
@@ -378,28 +418,28 @@
                             }
                         }
                     }
-                }
-            },
-            "parameters": [
-                {
-                    "name": "groupid",
-                    "in": "path",
-                    "description": "ID of the consumer group to which the subscribed consumer belongs.",
-                    "required": true,
-                    "schema": {
-                        "type": "string"
-                    }
                 },
-                {
-                    "name": "name",
-                    "in": "path",
-                    "description": "Name of the consumer to unsubscribe from topics.",
-                    "required": true,
-                    "schema": {
-                        "type": "string"
+                "parameters": [
+                    {
+                        "name": "groupid",
+                        "in": "path",
+                        "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
+                    },
+                    {
+                        "name": "name",
+                        "in": "path",
+                        "description": "Name of the consumer to unsubscribe from topics.",
+                        "required": true,
+                        "schema": {
+                            "type": "string"
+                        }
                     }
-                }
-            ]
+                ]
+            }
         },
         "/consumers/{groupid}/instances/{name}/positions": {
             "post": {
@@ -1546,7 +1586,7 @@
                         "type": "string"
                     },
                     "fetch.min.bytes": {
-                        "description": "Sets the minimum ammount of data, in bytes, for the consumer to receive. The broker waits until the data to send exceeds this amount.",
+                        "description": "Sets the minimum amount of data, in bytes, for the consumer to receive. The broker waits until the data to send exceeds this amount.",
                         "type": "integer"
                     },
                     "consumer.request.timeout.ms": {


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
index dea041d..cde2fb9 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
@@ -11,7 +11,7 @@ import io.strimzi.kafka.bridge.http.base.HttpBridgeITAbstract;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.utils.Urls;
 import io.vertx.core.MultiMap;
-import io.vertx.core.http.CaseInsensitiveHeaders;
+import io.vertx.core.http.impl.headers.HeadersMultiMap;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.client.HttpResponse;
@@ -319,7 +319,7 @@ public class ConsumerIT extends HttpBridgeITAbstract {
         String baseUri = "https://my-api-gateway-host:443/consumers/" + groupId + "/instances/" + name;
 
         // we have to use MultiMap because of https://github.com/vert-x3/vertx-web/issues/1383
-        MultiMap headers = new CaseInsensitiveHeaders();
+        MultiMap headers = new HeadersMultiMap();
         headers.add(FORWARDED, forwarded);
         headers.add(FORWARDED, forwarded2);
 
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
index c517fa8..5ca94c7 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
@@ -172,7 +172,7 @@ public class HttpCorsIT {
                     .putHeader("Origin", "https://strimzi.io")
                     .putHeader("Access-Control-Request-Method", "POST")
                     .send(ar -> context.verify(() -> {
-                        assertThat(ar.result().statusCode(), is(200));
+                        assertThat(ar.result().statusCode(), is(204));
                         assertThat(ar.result().getHeader("access-control-allow-origin"), is(origin));
                         assertThat(ar.result().getHeader("access-control-allow-headers"), is("access-control-allow-origin,content-length,x-forwarded-proto,x-forwarded-host,origin,x-requested-with,content-type,access-control-allow-methods,accept"));
                         List<String> list = Arrays.asList(ar.result().getHeader("access-control-allow-methods").split(","));
@@ -208,7 +208,7 @@ public class HttpCorsIT {
                     .putHeader("Origin", "https://strimzi.io")
                     .putHeader("Access-Control-Request-Method", "POST")
                     .send(ar -> context.verify(() -> {
-                        assertThat(ar.result().statusCode(), is(200));
+                        assertThat(ar.result().statusCode(), is(204));
                         assertThat(ar.result().getHeader("access-control-allow-origin"), is(origin));
                         assertThat(ar.result().getHeader("access-control-allow-headers"), is("access-control-allow-origin,content-length,x-forwarded-proto,x-forwarded-host,origin,x-requested-with,content-type,access-control-allow-methods,accept"));
                         List<String> list = Arrays.asList(ar.result().getHeader("access-control-allow-methods").split(","));
diff --git a/src/test/java/io/strimzi/kafka/bridge/utils/Urls.java b/src/test/java/io/strimzi/kafka/bridge/utils/Urls.java
index 6598633..14bf151 100644
--- a/src/test/java/io/strimzi/kafka/bridge/utils/Urls.java
+++ b/src/test/java/io/strimzi/kafka/bridge/utils/Urls.java
@@ -14,13 +14,13 @@ public class Urls {
 
     private static final String CONSUMERS_PATH = "/consumers/";
     private static final String INSTANCES_PATH =  "/instances/";
-    private static final String POSITIONS_BEGGINING_PATH =  "/positions/beginning";
+    private static final String POSITIONS_BEGINNING_PATH =  "/positions/beginning";
     private static final String POSITIONS_END_PATH =  "/positions/end";
     private static final String POSITIONS_PATH =  "/positions";
     private static final String SUBSCRIPTION_PATH =  "/subscription";
     private static final String TOPICS_PATH = "/topics/";
     private static final String PARTITIONS_PATH = "/partitions/";
-    private static final String ASSIGMENTS_PATH = "/assignments";
+    private static final String ASSIGNMENTS_PATH = "/assignments";
     private static final String OFFSETS_PATH = "/offsets";
     private static final String RECORDS_PATH = "/records";
 
@@ -33,7 +33,7 @@ public class Urls {
     }
 
     public static String consumerInstancePositionBeginning(String groupId, String name) {
-        return consumerInstance(groupId, name) + POSITIONS_BEGGINING_PATH;
+        return consumerInstance(groupId, name) + POSITIONS_BEGINNING_PATH;
     }
 
     public static String consumerInstancePositionEnd(String groupId, String name) {
@@ -49,7 +49,7 @@ public class Urls {
     }
 
     public static String consumerInstanceAssignments(String groupId, String name) {
-        return consumerInstance(groupId, name) + ASSIGMENTS_PATH;
+        return consumerInstance(groupId, name) + ASSIGNMENTS_PATH;
     }
 
     public static String consumerInstanceOffsets(String groupId, String name) {