Repository: strimzi/strimzi-kafka-bridge
Pull request #279 (2019-07-24 13:04:33)

Consumer name is prefixed by Bridge ID when available
https://github.com/strimzi/strimzi-kafka-bridge/issues/273

`name` neither `bridge.id` set.
```
{
    "instance_id": "kafka-bridge-consumer-a613d84b-4842-4a91-a5e3-ffe251b285e6",
    "base_uri": "http://localhost:8080/consumers/my-group/instances/kafka-bridge-consumer-a613d84b-4842-4a91-a5e3-ffe251b285e6"
}
```

`name` not set, `bridge.id` set to "my-bridge"
```
{
    "instance_id": "my-bridge-aafec785-0dab-4a49-830f-5339b664a2f5",
    "base_uri": "http://localhost:8080/consumers/my-group/instances/my-bridge-aafec785-0dab-4a49-830f-5339b664a2f5"
}
```


Patch:
diff --git a/.checkstyle/suppressions.xml b/.checkstyle/suppressions.xml
index 1334171..1c4d478 100644
--- a/.checkstyle/suppressions.xml
+++ b/.checkstyle/suppressions.xml
@@ -7,5 +7,6 @@
 <suppressions>
 
     <!-- Note that [/\\] must be used as the path separator for cross-platform support -->
-
+    <suppress checks="JavaNCSS"
+              files="io[/\\]strimzi[/\\]kafka[/\\]bridge[/\\]http[/\\]HttpBridgeTest.java"/>
 </suppressions>
\ No newline at end of file
diff --git a/bin/docker/kafka_bridge_config_generator.sh b/bin/docker/kafka_bridge_config_generator.sh
index 33193a3..17a1985 100755
--- a/bin/docker/kafka_bridge_config_generator.sh
+++ b/bin/docker/kafka_bridge_config_generator.sh
@@ -1,5 +1,11 @@
 #!/usr/bin/env bash
 
+BRIDGE_PROPERTIES=$(cat <<-EOF
+#Bridge configuration
+bridge.id=${KAFKA_BRIDGE_ID}
+EOF
+)
+
 SECURITY_PROTOCOL=PLAINTEXT
 
 if [ "$KAFKA_BRIDGE_TLS" = "true" ]; then
@@ -105,6 +111,8 @@ EOF
 
 # if http/amqp is disabled, do not print its configuration
 PROPERTIES=$(cat <<EOF
+$BRIDGE_PROPERTIES
+
 $KAFKA_PROPERTIES
 
 $PRODUCER_PROPERTIES
diff --git a/config/application.properties b/config/application.properties
index a3d3b07..035f7c2 100644
--- a/config/application.properties
+++ b/config/application.properties
@@ -1,3 +1,6 @@
+#Bridge related settings
+bridge.id=my-bridge
+
 #Apache Kafka common
 kafka.bootstrap.servers=localhost:9092
 
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
index 2ef1a9d..a66cd1a 100644
--- a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
+++ b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
@@ -18,13 +18,15 @@ public class BridgeConfig extends AbstractConfig {
 
     public static final String BRIDGE_CONFIG_PREFIX = "bridge.";
 
+    public static final String BRIDGE_ID = BRIDGE_CONFIG_PREFIX + "id";
+
     private KafkaConfig kafkaConfig;
     private AmqpConfig amqpConfig;
     private HttpConfig httpConfig;
 
     /**
      * Constructor
-     * 
+     *
      * @param config bridge common configuration parameters map
      * @param kafkaConfig Kafka related configuration
      * @param amqpConfig AMQP endpoint related configuration
@@ -71,7 +73,7 @@ public class BridgeConfig extends AbstractConfig {
 
         return new BridgeConfig(map.entrySet().stream()
                 .filter(e -> e.getKey().startsWith(BridgeConfig.BRIDGE_CONFIG_PREFIX))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)), 
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)),
                 kafkaConfig, amqpConfig, httpConfig);
     }
 
@@ -84,4 +86,12 @@ public class BridgeConfig extends AbstractConfig {
                 ",httpConfig=" + this.httpConfig +
                 ")";
     }
+
+    public String getBridgeID() {
+        if (config.get(BridgeConfig.BRIDGE_ID) == null) {
+            return null;
+        } else {
+            return config.get(BridgeConfig.BRIDGE_ID).toString();
+        }
+    }
 }
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 0c0a382..36d141c 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -73,7 +73,9 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         groupId = routingContext.pathParam("groupid");
 
         // if no name, a random one is assigned
-        this.name = bodyAsJson.getString("name", "kafka-bridge-consumer-" + UUID.randomUUID());
+        this.name = bodyAsJson.getString("name", bridgeConfig.getBridgeID() == null
+                ? "kafka-bridge-consumer-" + UUID.randomUUID()
+                : bridgeConfig.getBridgeID() + "-" + UUID.randomUUID());
 
         if (this.httpBridgeContext.getHttpSinkEndpoints().containsKey(this.name)) {
             HttpBridgeError error = new HttpBridgeError(
@@ -371,7 +373,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         try {
             bodyAsJson = routingContext.getBodyAsJson();
         } catch (DecodeException ex) {
-            
+
         }
         log.debug("[{}] Request: body = {}", routingContext.get("request-id"), bodyAsJson);
 
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
index 1356ccf..5437573 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -43,7 +43,7 @@ public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
 
     @Override
     public void open() {
-        this.name = "kafka-bridge-producer-" + UUID.randomUUID();
+        this.name = this.bridgeConfig.getBridgeID() == null ? "kafka-bridge-producer-" + UUID.randomUUID() : this.bridgeConfig.getBridgeID() + "-" + UUID.randomUUID();
         super.open();
     }
 
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
index a3d3b07..035f7c2 100644
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@@ -1,3 +1,6 @@
+#Bridge related settings
+bridge.id=my-bridge
+
 #Apache Kafka common
 kafka.bootstrap.servers=localhost:9092
 


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
index ab33fce..88dd54d 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -7,6 +7,7 @@ package io.strimzi.kafka.bridge.http;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
+import io.strimzi.kafka.bridge.HealthChecker;
 import io.strimzi.kafka.bridge.KafkaClusterTestBase;
 import io.strimzi.kafka.bridge.amqp.AmqpConfig;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
@@ -59,7 +60,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @ExtendWith(VertxExtension.class)
-@SuppressWarnings({"checkstyle:JavaNCSS"})
+@SuppressWarnings({"checkstyle:JavaNCSS", "checkstyle:ClassDataAbstractionCoupling"})
 class HttpBridgeTest extends KafkaClusterTestBase {
 
     private static final Logger log = LoggerFactory.getLogger(HttpBridgeTest.class);
@@ -112,6 +113,7 @@ class HttpBridgeTest extends KafkaClusterTestBase {
         this.vertx = Vertx.vertx();
         this.bridgeConfig = BridgeConfig.fromMap(config);
         this.httpBridge = new HttpBridge(this.bridgeConfig);
+        this.httpBridge.setHealthChecker(new HealthChecker());
 
         vertx.deployVerticle(this.httpBridge, context.succeeding(id -> context.completeNow()));
 
@@ -3396,4 +3398,92 @@ class HttpBridgeTest extends KafkaClusterTestBase {
         context.completeNow();
         assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
+
+    @Test
+    void readyTest(VertxTestContext context) throws InterruptedException {
+        int iterations = 5;
+        for (int i = 1; i <= iterations; i++) {
+            int l = i;
+            getRequest("/ready")
+                    .send(ar -> {
+                        context.verify(() -> {
+                            assertTrue(ar.succeeded());
+                            assertEquals(HttpResponseStatus.OK.code(), ar.result().statusCode());
+                        });
+                        if (l == iterations) {
+                            context.completeNow();
+                        }
+                    });
+            Thread.sleep(1000);
+        }
+    }
+
+    @Test
+    void healthyTest(VertxTestContext context) throws InterruptedException {
+        int iterations = 5;
+        for (int i = 1; i <= iterations; i++) {
+            int l = i;
+            getRequest("/healthy")
+                    .send(ar -> {
+                        context.verify(() -> {
+                            assertTrue(ar.succeeded());
+                            assertEquals(HttpResponseStatus.OK.code(), ar.result().statusCode());
+                        });
+                        if (l == iterations) {
+                            context.completeNow();
+                        }
+                    });
+            Thread.sleep(1000);
+        }
+    }
+
+    @Test
+    void openapiTest(VertxTestContext context) {
+        getRequest("/openapi")
+            .as(BodyCodec.jsonObject())
+            .send(ar -> {
+                context.verify(() -> {
+                    assertTrue(ar.succeeded());
+                    HttpResponse<JsonObject> response = ar.result();
+                    assertEquals(HttpResponseStatus.OK.code(), response.statusCode());
+                    JsonObject bridgeResponse = response.body();
+
+                    Map<String, Object> paths = bridgeResponse.getJsonObject("paths").getMap();
+                    // subscribe and unsubscribe are using the same endpoint but different methods
+                    int pathsSize = HttpOpenApiOperations.values().length - 1;
+                    assertEquals(pathsSize, paths.size());
+                    assertTrue(paths.containsKey("/consumers/{groupid}"));
+                    assertEquals(HttpOpenApiOperations.CREATE_CONSUMER.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/positions"));
+                    assertEquals(HttpOpenApiOperations.SEEK.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/positions").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/positions/beginning"));
+                    assertEquals(HttpOpenApiOperations.SEEK_TO_BEGINNING.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/positions/beginning").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/positions/end"));
+                    assertEquals(HttpOpenApiOperations.SEEK_TO_END.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/positions/end").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/subscription"));
+                    assertEquals(HttpOpenApiOperations.SUBSCRIBE.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("post").getString("operationId"));
+                    assertEquals(HttpOpenApiOperations.UNSUBSCRIBE.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("delete").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/assignments"));
+                    assertEquals(HttpOpenApiOperations.ASSIGN.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/assignments").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/records"));
+                    assertEquals(HttpOpenApiOperations.POLL.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/records").getJsonObject("get").getString("operationId"));
+                    assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/offsets"));
+                    assertEquals(HttpOpenApiOperations.COMMIT.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/offsets").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/topics/{topicname}"));
+                    assertEquals(HttpOpenApiOperations.SEND.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/topics/{topicname}").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/topics/{topicname}/partitions/{partitionid}"));
+                    assertEquals(HttpOpenApiOperations.SEND_TO_PARTITION.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/topics/{topicname}/partitions/{partitionid}").getJsonObject("post").getString("operationId"));
+                    assertTrue(paths.containsKey("/healthy"));
+                    assertEquals(HttpOpenApiOperations.HEALTHY.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/healthy").getJsonObject("get").getString("operationId"));
+                    assertTrue(paths.containsKey("/ready"));
+                    assertEquals(HttpOpenApiOperations.READY.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/ready").getJsonObject("get").getString("operationId"));
+                    assertTrue(paths.containsKey("/openapi"));
+                    assertEquals(HttpOpenApiOperations.OPENAPI.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/openapi").getJsonObject("get").getString("operationId"));
+                    assertFalse(paths.containsKey("/karel"));
+                    assertEquals(15, bridgeResponse.getJsonObject("definitions").getMap().size());
+                    assertEquals(4, bridgeResponse.getJsonArray("tags").size());
+                });
+                context.completeNow();
+            });
+    }
 }