Repository: strimzi/strimzi-kafka-bridge
Pull request #290 (2019-07-29 13:45:10)

Consumer timeout

scholzj: @ppatierno and @tombentley seem to be already reviewing this ... so I leave it up to them :-o

Patch:
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 1d175f1..2006aa9 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -33,6 +33,10 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
 /**
  * Main bridge class listening for connections and handling HTTP requests.
  */
@@ -54,6 +58,8 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
 
     private HealthChecker healthChecker;
 
+    private Map<String, Long> timestampMap = new HashMap<>();
+
     /**
      * Constructor
      *
@@ -77,6 +83,10 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
                                         .get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
                         );
 
+                        if (this.bridgeConfig.getHttpConfig().getConsumerTimeout() > -1) {
+                            startInactiveConsumerDeletionTimer(this.bridgeConfig.getHttpConfig().getConsumerTimeout());
+                        }
+
                         this.isReady = true;
                         startFuture.complete();
                     } else {
@@ -86,6 +96,26 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
                 });
     }
 
+    private void startInactiveConsumerDeletionTimer(Long timeout) {
+        Long timeoutInMs = timeout * 1000L;
+        vertx.setPeriodic(timeoutInMs / 2, ignore -> {
+            log.debug("Looking for stale consumers in {} entries", timestampMap.size());
+            Iterator it = timestampMap.entrySet().iterator();
+            while (it.hasNext()) {
+                Map.Entry<String, Long> item = (Map.Entry) it.next();
+                if (item.getValue() + timeoutInMs < System.currentTimeMillis()) {
+                    final SinkBridgeEndpoint deleteSinkEndpoint = this.httpBridgeContext.getHttpSinkEndpoints().get(item.getKey());
+                    if (deleteSinkEndpoint != null) {
+                        deleteSinkEndpoint.close();
+                        this.httpBridgeContext.getHttpSinkEndpoints().remove(item.getKey());
+                        log.warn("Consumer {} deleted after inactivity timeout ({}s).", item.getKey(), timeout);
+                        timestampMap.remove(item.getKey());
+                    }
+                }
+            }
+        });
+    }
+
     @Override
     public void start(Future<Void> startFuture) throws Exception {
 
@@ -139,7 +169,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         });
         this.httpBridgeContext.getHttpSinkEndpoints().clear();
 
-        //prducer 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) -> {
@@ -200,6 +230,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
 
             sink.handle(new HttpEndpoint(routingContext), consumerId -> {
                 httpBridgeContext.getHttpSinkEndpoints().put(consumerId.toString(), sink);
+                timestampMap.put(consumerId.toString(), System.currentTimeMillis());
             });
         } catch (Exception ex) {
             sink.close();
@@ -222,6 +253,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
             deleteSinkEndpoint.handle(new HttpEndpoint(routingContext));
 
             this.httpBridgeContext.getHttpSinkEndpoints().remove(deleteInstanceID);
+            timestampMap.remove(deleteInstanceID);
         } else {
             HttpBridgeError error = new HttpBridgeError(
                     HttpResponseStatus.NOT_FOUND.code(),
@@ -283,6 +315,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         final SinkBridgeEndpoint sinkEndpoint = this.httpBridgeContext.getHttpSinkEndpoints().get(instanceId);
 
         if (sinkEndpoint != null) {
+            timestampMap.replace(instanceId, System.currentTimeMillis());
             sinkEndpoint.handle(new HttpEndpoint(routingContext));
         } else {
             HttpBridgeError error = new HttpBridgeError(
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
index 5c1fc82..0b3975c 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
@@ -20,10 +20,12 @@ public class HttpConfig extends AbstractConfig {
     public static final String HTTP_ENABLED = HTTP_CONFIG_PREFIX + "enabled";
     public static final String HTTP_HOST = HTTP_CONFIG_PREFIX + "host";
     public static final String HTTP_PORT = HTTP_CONFIG_PREFIX + "port";
+    public static final String HTTP_CONSUMER_TIMEOUT = HTTP_CONFIG_PREFIX + "timeoutSeconds";
 
     public static final boolean DEFAULT_HTTP_ENABLED = true;
     public static final String DEFAULT_HOST = "0.0.0.0";
     public static final int DEFAULT_PORT = 8080;
+    public static final long DEFAULT_CONSUMER_TIMEOUT = -1L;
 
     /**
      * Constructor
@@ -55,6 +57,13 @@ public class HttpConfig extends AbstractConfig {
         return Integer.parseInt(this.config.getOrDefault(HTTP_PORT, DEFAULT_PORT).toString());
     }
 
+    /**
+     * @return the timeout for closing inactive consumer
+     */
+    public long getConsumerTimeout() {
+        return Long.parseLong(this.config.getOrDefault(HTTP_CONSUMER_TIMEOUT, DEFAULT_CONSUMER_TIMEOUT).toString());
+    }
+
     /**
      * Loads HTTP related configuration parameters from a related map
      *


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 ddc46cf..b2fbc6d 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -67,10 +67,13 @@ class HttpBridgeTest extends KafkaClusterTestBase {
 
     private static Map<String, Object> config = new HashMap<>();
 
+    private static long timeout = 5L;
+
     static {
         config.put(AmqpConfig.AMQP_ENABLED, true);
         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);
     }
 
     private static final String BRIDGE_HOST = "127.0.0.1";
@@ -3588,4 +3591,53 @@ class HttpBridgeTest extends KafkaClusterTestBase {
                 context.completeNow();
             });
     }
+
+    @Test
+    void consumerDeletedAfterInactivity(VertxTestContext context) {
+        String name = "my-kafka-consumer";
+        String groupId = "my-group";
+
+        String baseUri = "http://" + BRIDGE_HOST + ":" + BRIDGE_PORT + "/consumers/" + groupId + "/instances/" + name;
+
+        JsonObject consumerJson = new JsonObject();
+        consumerJson.put("name", name);
+        consumerJson.put("format", "json");
+
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
+        postRequest("/consumers/" + groupId)
+                .putHeader("Content-length", String.valueOf(consumerJson.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
+                .as(BodyCodec.jsonObject())
+                .sendJsonObject(consumerJson, 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");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+
+                        vertx.setTimer(timeout * 2 * 1000L, timeouted -> {
+                            CompletableFuture<Boolean> delete = new CompletableFuture<>();
+                            // consumer deletion
+                            deleteRequest(baseUri)
+                                    .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
+                                    .as(BodyCodec.jsonObject())
+                                    .send(consumerDeletedResponse -> {
+                                        context.verify(() -> assertTrue(ar.succeeded()));
+
+                                        HttpResponse<JsonObject> deletionResponse = consumerDeletedResponse.result();
+                                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), deletionResponse.statusCode());
+                                        assertEquals("The specified consumer instance was not found.", deletionResponse.body().getString("message"));
+
+                                        delete.complete(true);
+                                        context.completeNow();
+                                    });
+                        });
+                    });
+                    create.complete(true);
+                });
+    }
 }