commit 2fe60855f97c8e0cf51ab6133ce74c46355ef46f
author Rikuo Takahama <mas.null19@gmail.com> Sat Sep 07 00:36:07 2019 +0900
committer Matteo Merli <mmerli@apache.org> Fri Sep 06 08:36:07 2019 -0700
tree b526b0d43046742dd019bb8475145a1e2dcc0adc
parent b0b1ff8d6577f16f51776d2cc407080a8ec7ccb4

Make client keepalive interval configurable on pulsar-client-kafka (#5131)

pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarClientKafkaConfig.java

diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarClientKafkaConfig.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarClientKafkaConfig.java
index c38d93d..8ba5ede 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarClientKafkaConfig.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarClientKafkaConfig.java
@@ -48,6 +48,8 @@
     public static final String CONCURRENT_LOOKUP_REQUESTS = "pulsar.concurrent.lookup.requests";
     public static final String MAX_NUMBER_OF_REJECTED_REQUESTS_PER_CONNECTION = "pulsar.max.number.rejected.request.per.connection";
 
+    public static final String KEEPALIVE_INTERVAL_MS = "pulsar.keepalive.interval.ms";
+
     public static ClientBuilder getClientBuilder(Properties properties) {
         ClientBuilder clientBuilder = PulsarClient.builder();
 
@@ -112,6 +114,11 @@
                     Integer.parseInt(properties.getProperty(MAX_NUMBER_OF_REJECTED_REQUESTS_PER_CONNECTION)));
         }
 
+        if (properties.containsKey(KEEPALIVE_INTERVAL_MS)) {
+            clientBuilder.keepAliveInterval(Integer.parseInt(properties.getProperty(KEEPALIVE_INTERVAL_MS)),
+                    TimeUnit.MILLISECONDS);
+        }
+
         return clientBuilder;
     }
 }