[GitHub] sijie closed pull request #3308: Add heartbeat interval for kafka connector

[GitBox <gi...@apache.org>]

sijie closed pull request #3308: Add heartbeat interval for kafka connector
URL: https://github.com/apache/pulsar/pull/3308
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

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 6c6d1dabe8..dc3a733bea 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 @@
     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 void open(Map<String, Object> config, SourceContext sourceContext) throws
             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 void open(Map<String, Object> config, SourceContext sourceContext) throws
         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 c60c9debe7..6bf7d66320 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
@@ -69,6 +69,12 @@
         help =
             "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 =
diff --git a/site2/docs/io-kafka.md b/site2/docs/io-kafka.md
index 2ae53be969..5eab2c1f5f 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 cee17b1451..f5854e7e6f 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 KafkaSourceTester() {
         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");
     }


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services