[pulsar] branch master updated: Add heartbeat interval for kafka connector (#3308)

[si...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new b79bb92  Add heartbeat interval for kafka connector (#3308)
b79bb92 is described below

commit b79bb9211d739c3e850641000a9cbadf6e785aa7
Author: Ke Li <li...@163.com>
AuthorDate: Mon Jan 7 04:52:37 2019 +0800

    Add heartbeat interval for kafka connector (#3308)
    
    ### Motivation
    
    The default value of ```heartbeat.interval.ms``` is 3000 in latest kafka consumer config, which cannot be customized in kafka source config now.  When user set ```sessionTimeoutMs``` to a value less than or equal to 3000, an exception will thrown. So we'd better to provide the ability to support this option.
    
    Closes #3293
---
 .../main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java  | 7 ++++++-
 .../main/java/org/apache/pulsar/io/kafka/KafkaSourceConfig.java    | 6 ++++++
 site2/docs/io-kafka.md                                             | 1 +
 .../org/apache/pulsar/tests/integration/io/KafkaSourceTester.java  | 1 +
 4 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
index 6c6d1da..dc3a733 100644
--- a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
+++ b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
@@ -49,7 +49,7 @@ public abstract class KafkaAbstractSource<V> extends PushSource<V> {
     private Consumer<String, byte[]> consumer;
     private Properties props;
     private KafkaSourceConfig kafkaSourceConfig;
-    Thread runnerThread;
+    private Thread runnerThread;
 
     @Override
     public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
@@ -69,6 +69,10 @@ public abstract class KafkaAbstractSource<V> extends PushSource<V> {
             throw new IllegalArgumentException("Invalid Kafka Consumer sessionTimeoutMs : "
                 + kafkaSourceConfig.getSessionTimeoutMs());
         }
+        if (kafkaSourceConfig.getHeartbeatIntervalMs() <= 0) {
+            throw new IllegalArgumentException("Invalid Kafka Consumer heartbeatIntervalMs : "
+                + kafkaSourceConfig.getHeartbeatIntervalMs());
+        }
 
         props = new Properties();
 
@@ -77,6 +81,7 @@ public abstract class KafkaAbstractSource<V> extends PushSource<V> {
         props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, String.valueOf(kafkaSourceConfig.getFetchMinBytes()));
         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(kafkaSourceConfig.getAutoCommitIntervalMs()));
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, String.valueOf(kafkaSourceConfig.getSessionTimeoutMs()));
+        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, String.valueOf(kafkaSourceConfig.getHeartbeatIntervalMs()));
         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaSourceConfig.getKeyDeserializationClass());
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaSourceConfig.getValueDeserializationClass());
diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSourceConfig.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSourceConfig.java
index c60c9de..6bf7d66 100644
--- a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSourceConfig.java
+++ b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSourceConfig.java
@@ -70,6 +70,12 @@ public class KafkaSourceConfig implements Serializable {
             "The timeout used to detect failures when using Kafka's group management facilities.")
     private long sessionTimeoutMs = 30000L;
     @FieldDoc(
+        defaultValue = "3000",
+        help =
+            "The interval between heartbeats to the consumer when using Kafka's group management facilities. "
+                + "The value must be lower than session timeout.")
+    private long heartbeatIntervalMs = 3000L;
+    @FieldDoc(
         defaultValue = "true",
         help =
             "If true the consumer's offset will be periodically committed in the background.")
diff --git a/site2/docs/io-kafka.md b/site2/docs/io-kafka.md
index 2ae53be..5eab2c1 100644
--- a/site2/docs/io-kafka.md
+++ b/site2/docs/io-kafka.md
@@ -18,6 +18,7 @@ to a Pulsar topic.
 | fetchMinBytes | `false` | `null` | Minimum bytes expected for each fetch response. |
 | autoCommitEnabled | `false` | `false` | If true, periodically commit to ZooKeeper the offset of messages already fetched by the consumer. This committed offset will be used when the process fails as the position from which the new consumer will begin. | 
 | autoCommitIntervalMs | `false` | `null` | The frequency in ms that the consumer offsets are committed to zookeeper. |
+| heartbeatIntervalMs | `false` | `3000` | The interval between heartbeats to the consumer when using Kafka's group management facilities. |
 | sessionTimeoutMs | `false` | `null` | The timeout used to detect consumer failures when using Kafka's group management facility. |
 | topic | `true` | `null` | Topic name to receive records from Kafka |
 | keySerializerClass | false | org.apache.kafka.common.serialization.StringSerializer | Serializer class for key that implements the org.apache.kafka.common.serialization.Serializer interface. |
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java
index cee17b1..f5854e7 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java
@@ -61,6 +61,7 @@ public class KafkaSourceTester extends SourceTester<KafkaContainer> {
         sourceConfig.put("fetchMinBytes", 1L);
         sourceConfig.put("autoCommitIntervalMs", 10L);
         sourceConfig.put("sessionTimeoutMs", 10000L);
+        sourceConfig.put("heartbeatIntervalMs", 5000L);
         sourceConfig.put("topic", kafkaTopicName);
         sourceConfig.put("valueDeserializationClass", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
     }