Repository: strimzi/strimzi-kafka-bridge
Pull request #490 (2020-11-09 11:52:41)

Assignment test + NPE fix
Signed-off-by: Stanislav Knot <sknot@redhat.com>

Assigment endpoint does not have any tests. 
Trying to access `subscribe` and `assign` endpoints ends up with NPE. After fix of NPE, it does not return 409 (Conflict).

Fixes https://github.com/strimzi/strimzi-kafka-bridge/issues/489


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/SinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
index d0b0a7b..481ed2d 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
@@ -69,6 +69,10 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
     protected List<SinkTopicSubscription> topicSubscriptions;
     protected Pattern topicSubscriptionsPattern;
 
+    protected boolean subscribed;
+    protected boolean assigned;
+
+
     private int recordIndex;
     private int batchSize;
 
@@ -115,6 +119,8 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.format = format;
         this.keyDeserializer = keyDeserializer;
         this.valueDeserializer = valueDeserializer;
+        this.subscribed = false;
+        this.assigned = false;
     }
 
     @Override
@@ -191,6 +197,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.shouldAttachSubscriberHandler = shouldAttachHandler;
 
         log.info("Subscribe to topics {}", this.topicSubscriptions);
+        this.subscribed = true;
         this.setPartitionsAssignmentHandlers();
 
         Set<String> topics = this.topicSubscriptions.stream().map(ts -> ts.getTopic()).collect(Collectors.toSet());
@@ -198,12 +205,14 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
     }
 
     /**
-     * Unubscribe all the topics which the consumer currently subscribes
+     * Unsubscribe all the topics which the consumer currently subscribes
      */
     protected void unsubscribe() {
         log.info("Unsubscribe from topics {}", this.topicSubscriptions);
         topicSubscriptions.clear();
         topicSubscriptionsPattern = null;
+        this.subscribed = false;
+        this.assigned = false;
         this.consumer.unsubscribe(this::unsubscribeHandler);
     }
 
@@ -228,6 +237,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
 
         log.info("Subscribe to topics with pattern {}", pattern);
         this.setPartitionsAssignmentHandlers();
+        this.subscribed = true;
         this.consumer.subscribe(pattern, this::subscribeHandler);
     }
 
@@ -276,6 +286,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.shouldAttachSubscriberHandler = shouldAttachHandler;
 
         log.info("Assigning to topics partitions {}", this.topicSubscriptions);
+        this.assigned = true;
         this.partitionsAssignmentAndSeek();
     }
 
@@ -360,6 +371,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
             Optional<PartitionInfo> requestedPartitionInfo =
                     availablePartitions.stream()
                             .filter(p -> p.getTopic().equals(topicSubscription.getTopic()) &&
+                                    topicSubscription.getPartition() != null &&
                                     p.getPartition() == topicSubscription.getPartition())
                             .findFirst();
 
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..8d832b3 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -355,6 +355,14 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
     }
 
     private void doAssign(RoutingContext routingContext, JsonObject bodyAsJson) {
+        if (subscribed) {
+            HttpBridgeError error = new HttpBridgeError(
+                    HttpResponseStatus.CONFLICT.code(), "Subscriptions to topics, partitions, and patterns are mutually exclusive."
+            );
+            HttpUtils.sendResponse(routingContext, HttpResponseStatus.CONFLICT.code(),
+                    BridgeContentType.KAFKA_JSON, error.toJson().toBuffer());
+            return;
+        }
         JsonArray partitionsList = bodyAsJson.getJsonArray("partitions");
         this.topicSubscriptions.addAll(
                 partitionsList.stream()
@@ -374,7 +382,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
 
     private void doSubscribe(RoutingContext routingContext, JsonObject bodyAsJson) {
         // cannot specify both topics list and topic pattern
-        if (bodyAsJson.containsKey("topics") && bodyAsJson.containsKey("topic_pattern")) {
+        if ((bodyAsJson.containsKey("topics") && bodyAsJson.containsKey("topic_pattern")) || assigned) {
             HttpBridgeError error = new HttpBridgeError(
                     HttpResponseStatus.CONFLICT.code(),
                     "Subscriptions to topics, partitions, and patterns are mutually exclusive."
@@ -483,6 +491,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 +499,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/ConsumerSubscriptionIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java
index 9623ea1..0bba689 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java
@@ -224,7 +224,7 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
 
         consumerService()
                 .createConsumer(context, groupId, json)
-                .subscribeConsumer(context, groupId, name, topic, topic2);
+                .subscribeConsumer(context, groupId, name, topicsRoot);
 
         // poll to subscribe
         CompletableFuture<Boolean> consume = new CompletableFuture<>();
@@ -292,6 +292,60 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                     consume.complete(true);
                 });
         consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        consumerService()
+            .deleteConsumer(context, groupId, name);
+        context.completeNow();
+    }
+
+    @Test
+    void assignAfterSubscriptionTest(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
+        String topic = "subscribe-and-assign-topic";
+
+        adminClientFacade.createTopic(topic, 4, 1);
+
+        assertThat(adminClientFacade.listTopic().size(), is(1));
+
+        String name = "my-kafka-consumer-assign";
+
+        JsonObject json = new JsonObject();
+        json.put("name", name);
+        json.put("format", "json");
+
+        JsonObject partitionsRoot = new JsonObject();
+        JsonArray partitions = new JsonArray();
+        JsonObject part0 = new JsonObject();
+        part0.put("topic", topic);
+        part0.put("partition", 0);
+
+        JsonObject part1 = new JsonObject();
+        part1.put("topic", topic);
+        part1.put("partition", 1);
+        partitions.add(part0);
+        partitions.add(part1);
+
+        partitionsRoot.put("partitions", partitions);
+
+        consumerService()
+                .createConsumer(context, groupId, json)
+                .subscribeConsumer(context, groupId, name, topic);
+
+        CompletableFuture<Boolean> assignCF = new CompletableFuture<>();
+        consumerService()
+                .assignRequest(groupId, name, partitionsRoot)
+                .sendJsonObject(partitionsRoot, ar -> {
+                    context.verify(() -> {
+                        assertThat(ar.succeeded(), is(true));
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertThat(response.statusCode(), is(HttpResponseStatus.CONFLICT.code()));
+                        assertThat(error.getCode(), is(HttpResponseStatus.CONFLICT.code()));
+                        assertThat(error.getMessage(), is("Subscriptions to topics, partitions, and patterns are mutually exclusive."));
+                    });
+                    assignCF.complete(true);
+                });
+
+        assignCF.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+
         consumerService()
                 .deleteConsumer(context, groupId, name);
         context.completeNow();
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/http/services/ConsumerService.java b/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
index 681dd99..3e4800e 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
@@ -93,6 +93,13 @@ public class ConsumerService extends BaseService {
                 .as(BodyCodec.jsonObject());
     }
 
+    public HttpRequest<JsonObject> assignRequest(String groupId, String name, JsonObject json) {
+        return postRequest(Urls.consumerInstanceAssignments(groupId, name))
+                .putHeader(CONTENT_LENGTH.toString(), String.valueOf(json.toBuffer().length()))
+                .putHeader(CONTENT_TYPE.toString(), BridgeContentType.KAFKA_JSON)
+                .as(BodyCodec.jsonObject());
+    }
+
     public HttpRequest<JsonObject> offsetsRequest(String groupId, String name) {
         return postRequest(Urls.consumerInstanceOffsets(groupId, name));
     }
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) {