Test for consumer name prefix (bridge ID)

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

@@ -161,24 +161,13 @@ public void stop(Future<Void> stopFuture) throws Exception {
 
         this.isReady = false;
 
-        //Consumers cleanup
-        this.httpBridgeContext.getHttpSinkEndpoints().forEach((consumerId, httpEndpoint) -> {
-            if (httpEndpoint != null) {
-                httpEndpoint.close();
-            }
-        });
-        this.httpBridgeContext.getHttpSinkEndpoints().clear();
+        // Consumers cleanup
+        this.httpBridgeContext.closeAllSinkBridgeEndpoints();
 
-        //producer cleanup
+        // producer cleanup
         // for each connection, we have to close the connection itself but before that
         // all the sink/source endpoints (so the related links inside each of them)
-        this.httpBridgeContext.getHttpSourceEndpoints().forEach((connection, endpoint) -> {
-
-            if (endpoint != null) {
-                endpoint.close();
-            }
-        });
-        this.httpBridgeContext.getHttpSourceEndpoints().clear();
+        this.httpBridgeContext.closeAllSourceBridgeEndpoints();
 
         if (this.httpServer != null) {
 

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

@@ -56,4 +56,20 @@ public void setOpenApiOperation(HttpOpenApiOperations openApiOperation) {
     public HttpOpenApiOperations getOpenApiOperation() {
         return this.openApiOperation;
     }
+
+    public void closeAllSinkBridgeEndpoints() {
+        for (Map.Entry<String, SinkBridgeEndpoint> sink: getHttpSinkEndpoints().entrySet()) {
+            if (sink.getValue() != null)
+                sink.getValue().close();
+        }
+        getHttpSinkEndpoints().clear();
+    }
+
+    public void closeAllSourceBridgeEndpoints() {
+        for (Map.Entry<HttpConnection, SourceBridgeEndpoint> source: getHttpSourceEndpoints().entrySet()) {
+            if (source.getValue() != null)
+                source.getValue().close();
+        }
+        getHttpSourceEndpoints().clear();
+    }
 }

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

@@ -74,6 +74,7 @@ class HttpBridgeTest extends KafkaClusterTestBase {
         config.put(KafkaConfig.KAFKA_CONFIG_PREFIX + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         config.put(KafkaConsumerConfig.KAFKA_CONSUMER_CONFIG_PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         config.put(HttpConfig.HTTP_CONSUMER_TIMEOUT, timeout);
+        config.put(BridgeConfig.BRIDGE_ID, "my-bridge");
     }
 
     private static final String BRIDGE_HOST = "127.0.0.1";
@@ -3640,4 +3641,27 @@ void consumerDeletedAfterInactivity(VertxTestContext context) {
                     create.complete(true);
                 });
     }
+
+    @Test
+    void createConsumerWithGeneratedName(VertxTestContext context) {
+        String groupId = "my-group";
+
+        JsonObject json = new JsonObject();
+
+        postRequest("/consumers/" + groupId)
+                .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
+                .as(BodyCodec.jsonObject())
+                .sendJsonObject(json, ar -> {
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.OK.code(), response.statusCode());
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        assertTrue(consumerInstanceId.startsWith(config.get(BridgeConfig.BRIDGE_ID).toString()));
+                    });
+                    context.completeNow();
+                });
+    }
 }