This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 52efe7a40b issue #7519 : Kafka Consumer stop when idle (#7524)
52efe7a40b is described below
commit 52efe7a40b7c323345f8fbe325d51a164a8fca5e
Author: Matt Casters <[email protected]>
AuthorDate: Wed Jul 15 13:40:10 2026 +0200
issue #7519 : Kafka Consumer stop when idle (#7524)
* issue #7519 : Kafka Consumer stop when idle
Add stopWhenIdle and maxIdleTimeMs options so the consumer can drain a
topic and exit after a configurable idle period (batch-style use cases).
* issue #7519 : fix Kafka IT bootstrap and idle assignment race
Harden BOOTSTRAP_SERVERS for the single-JVM runner, wait for Kafka
health in docker, ignore idle time until partitions are assigned, and
convert basic/mapping ITs to stop-when-idle.
---
.../pages/pipeline/transforms/kafkaconsumer.adoc | 14 +-
.../0001-kafka-consumer-read-record-basic.hpl | 30 +---
.../0002-kafka-consumer-read-record-mapping.hpl | 29 +---
.../0003-kafka-consumer-called-subpipeline.hpl | 112 ++++++++++++++
...=> 0003-kafka-consumer-read-stop-when-idle.hpl} | 36 +----
.../kafka/main-0003-kafka-test-stop-when-idle.hwf | 171 +++++++++++++++++++++
... => prepare-kafka-test-0003-stop-when-idle.hpl} | 105 +++++--------
.../kafka/consumer/KafkaConsumerInput.java | 30 +++-
.../kafka/consumer/KafkaConsumerInputData.java | 3 +
.../kafka/consumer/KafkaConsumerInputDialog.java | 37 ++++-
.../kafka/consumer/KafkaConsumerInputMeta.java | 35 ++++-
.../consumer/messages/messages_en_US.properties | 4 +
.../kafka/consumer/KafkaConsumerInputMetaTest.java | 25 +++
13 files changed, 472 insertions(+), 159 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
index 28ee201c4a..a388360e51 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaconsumer.adoc
@@ -16,7 +16,7 @@ under the License.
////
:documentationPath: /pipeline/transforms/
:language: en_US
-:description: The Kafka Consumer transform continuously pulls data from Kafka.
It runs a sub-pipeline that executes according to message batch size or
duration.
+:description: The Kafka Consumer transform pulls data from Kafka. It runs a
sub-pipeline that executes according to message batch size or duration, and can
optionally stop when idle.
= image:transforms/icons/KafkaConsumerInput.svg[Kafka Consumer transform Icon,
role="image-doc-icon"] Kafka Consumer
@@ -40,9 +40,11 @@ Kafka scales topic consumption by distributing partitions
among a consumer group
A consumer group is a set of consumers sharing a common group identifier.
-Since the Kafka Consumer transform continuously ingests streaming data, you
may want to use the Abort transform in your parent or sub-pipeline to stop
consuming records from Kafka for specific workflows.
+By default the Kafka Consumer transform continuously ingests streaming data.
+To stop after the topic has been drained, enable *Stop when idle* on the Batch
tab (with an optional max idle time).
+You can also use the Abort transform in your parent or sub-pipeline to stop
consuming records for other conditions.
-For example, you can run the parent pipeline on a timed schedule, or abort the
sub-pipeline if sensor data exceeds a preset range.
+For example, you can run the parent pipeline on a timed schedule, drain a
topic once per hour with stop-when-idle, or abort the sub-pipeline if sensor
data exceeds a preset range.
|
== Supported Engines
@@ -99,6 +101,12 @@ If set to a value of ‘0’, then Number of records triggers
consumption.
|Number of records|Specify a number.
After every ‘X’ number of records, the specified pipeline will be executed and
these ‘X’ records will be passed to the pipeline.
If set to a value of ‘0’ then Duration triggers consumption.
+|Stop when idle|When enabled, the consumer stops gracefully after no records
are received for the configured max idle time.
+This supports batch-style use cases such as draining a topic periodically
instead of running forever.
+While this option is enabled, the consumer polls with a short (100 ms) timeout
so idle time can be measured accurately.
+|Max idle time (ms)|The maximum time in milliseconds to wait without receiving
records before stopping when *Stop when idle* is enabled.
+Defaults to 500.
+Supports variables.
|Offset management a|Choose when to commit
* when record read
diff --git a/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
b/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
index f10879691b..c6eac958df 100644
--- a/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
+++ b/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
@@ -46,11 +46,6 @@ limitations under the License.
<to>Log Output</to>
<enabled>Y</enabled>
</hop>
- <hop>
- <from>Log Output</from>
- <to>Stop Kafka Processing</to>
- <enabled>Y</enabled>
- </hop>
</order>
<transform>
<name>Kafka Consumer</name>
@@ -67,8 +62,10 @@ limitations under the License.
<consumerGroup/>
<pipelinePath>${PROJECT_HOME}/0001-kafka-consumer-called-subpipeline.hpl</pipelinePath>
<subTransform>Out record</subTransform>
- <batchSize>1</batchSize>
+ <batchSize>10</batchSize>
<batchDuration>1000</batchDuration>
+ <stopWhenIdle>Y</stopWhenIdle>
+ <maxIdleTimeMs>5000</maxIdleTimeMs>
<directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
<AUTO_COMMIT>N</AUTO_COMMIT>
<OutputField kafkaName="key" type="String">Key</OutputField>
@@ -118,27 +115,6 @@ limitations under the License.
<yloc>112</yloc>
</GUI>
</transform>
- <transform>
- <name>Stop Kafka Processing</name>
- <type>Abort</type>
- <description/>
- <distribute>Y</distribute>
- <custom_distribution/>
- <copies>1</copies>
- <partitioning>
- <method>none</method>
- <schema_name/>
- </partitioning>
- <abort_option>SAFE_STOP</abort_option>
- <always_log_rows>Y</always_log_rows>
- <message>Kafka consumer stopped</message>
- <row_threshold>0</row_threshold>
- <attributes/>
- <GUI>
- <xloc>672</xloc>
- <yloc>112</yloc>
- </GUI>
- </transform>
<transform_error_handling>
</transform_error_handling>
<attributes/>
diff --git
a/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
b/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
index 93d136cc7d..546080a909 100644
--- a/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
+++ b/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
@@ -61,8 +61,10 @@ limitations under the License.
<pipelinePath>${PROJECT_HOME}/0002-kafka-consumer-called-subpipeline-with-mapping.hpl</pipelinePath>
<executionInformationLocation/>
<executionDataProfile/>
- <batchSize>1</batchSize>
+ <batchSize>10</batchSize>
<batchDuration>1000</batchDuration>
+ <stopWhenIdle>Y</stopWhenIdle>
+ <maxIdleTimeMs>5000</maxIdleTimeMs>
<subTransform>Out record</subTransform>
<directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
<topic>hop-test-0002</topic>
@@ -128,37 +130,12 @@ limitations under the License.
</partitioning>
<attributes/>
</transform>
- <transform>
- <type>Abort</type>
- <name>Stop Kafka Processing</name>
- <row_threshold>0</row_threshold>
- <message>Kafka consumer stopped</message>
- <always_log_rows>Y</always_log_rows>
- <abort_option>SAFE_STOP</abort_option>
- <distribute>Y</distribute>
- <copies>1</copies>
- <GUI>
- <xloc>672</xloc>
- <yloc>112</yloc>
- </GUI>
- <description/>
- <partitioning>
- <method>none</method>
- <schema_name/>
- </partitioning>
- <attributes/>
- </transform>
<order>
<hop>
<from>Kafka Consumer</from>
<to>Log Output</to>
<enabled>Y</enabled>
</hop>
- <hop>
- <from>Log Output</from>
- <to>Stop Kafka Processing</to>
- <enabled>Y</enabled>
- </hop>
</order>
<notepads/>
<attributes/>
diff --git a/integration-tests/kafka/0003-kafka-consumer-called-subpipeline.hpl
b/integration-tests/kafka/0003-kafka-consumer-called-subpipeline.hpl
new file mode 100644
index 0000000000..4d0117a1f4
--- /dev/null
+++ b/integration-tests/kafka/0003-kafka-consumer-called-subpipeline.hpl
@@ -0,0 +1,112 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>0003-kafka-consumer-called-subpipeline</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2021/12/21 09:43:31.432</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2021/12/21 09:43:31.432</modified_date>
+ <key_for_session_key>H4sIAAAAAAAAAAMAAAAAAAAAAAA=</key_for_session_key>
+ <is_key_private>N</is_key_private>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Injector</from>
+ <to>Out record</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Out record</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>384</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Injector</name>
+ <type>Injector</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>key</name>
+ <type>String</type>
+ </field>
+ <field>
+ <name>message</name>
+ <type>String</type>
+ </field>
+ <field>
+ <name>topic</name>
+ <type>String</type>
+ </field>
+ <field>
+ <name>partition</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <name>offset</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <name>timestamp</name>
+ <type>Integer</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>240</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
b/integration-tests/kafka/0003-kafka-consumer-read-stop-when-idle.hpl
similarity index 82%
copy from integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
copy to integration-tests/kafka/0003-kafka-consumer-read-stop-when-idle.hpl
index f10879691b..2c7f79b7c4 100644
--- a/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
+++ b/integration-tests/kafka/0003-kafka-consumer-read-stop-when-idle.hpl
@@ -19,7 +19,7 @@ limitations under the License.
-->
<pipeline>
<info>
- <name>0001-kafka-consumer-read-record-basic</name>
+ <name>0003-kafka-consumer-read-stop-when-idle</name>
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
@@ -46,11 +46,6 @@ limitations under the License.
<to>Log Output</to>
<enabled>Y</enabled>
</hop>
- <hop>
- <from>Log Output</from>
- <to>Stop Kafka Processing</to>
- <enabled>Y</enabled>
- </hop>
</order>
<transform>
<name>Kafka Consumer</name>
@@ -63,12 +58,14 @@ limitations under the License.
<method>none</method>
<schema_name/>
</partitioning>
- <topic>hop-test</topic>
+ <topic>hop-test-idle</topic>
<consumerGroup/>
-
<pipelinePath>${PROJECT_HOME}/0001-kafka-consumer-called-subpipeline.hpl</pipelinePath>
+
<pipelinePath>${PROJECT_HOME}/0003-kafka-consumer-called-subpipeline.hpl</pipelinePath>
<subTransform>Out record</subTransform>
- <batchSize>1</batchSize>
+ <batchSize>10</batchSize>
<batchDuration>1000</batchDuration>
+ <stopWhenIdle>Y</stopWhenIdle>
+ <maxIdleTimeMs>5000</maxIdleTimeMs>
<directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
<AUTO_COMMIT>N</AUTO_COMMIT>
<OutputField kafkaName="key" type="String">Key</OutputField>
@@ -118,27 +115,6 @@ limitations under the License.
<yloc>112</yloc>
</GUI>
</transform>
- <transform>
- <name>Stop Kafka Processing</name>
- <type>Abort</type>
- <description/>
- <distribute>Y</distribute>
- <custom_distribution/>
- <copies>1</copies>
- <partitioning>
- <method>none</method>
- <schema_name/>
- </partitioning>
- <abort_option>SAFE_STOP</abort_option>
- <always_log_rows>Y</always_log_rows>
- <message>Kafka consumer stopped</message>
- <row_threshold>0</row_threshold>
- <attributes/>
- <GUI>
- <xloc>672</xloc>
- <yloc>112</yloc>
- </GUI>
- </transform>
<transform_error_handling>
</transform_error_handling>
<attributes/>
diff --git a/integration-tests/kafka/main-0003-kafka-test-stop-when-idle.hwf
b/integration-tests/kafka/main-0003-kafka-test-stop-when-idle.hwf
new file mode 100644
index 0000000000..f4aa237a77
--- /dev/null
+++ b/integration-tests/kafka/main-0003-kafka-test-stop-when-idle.hwf
@@ -0,0 +1,171 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<workflow>
+ <name>main-0003-kafka-test-stop-when-idle</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <created_user>-</created_user>
+ <modified_user>-</modified_user>
+ <created_date>2021/12/30 10:42:13.629</created_date>
+ <modified_date>2021/12/30 10:42:13.629</modified_date>
+ <workflow_version/>
+ <parameters>
+ <parameter>
+ <name>BOOTSTRAP_SERVERS</name>
+ <description/>
+ <default_value>kafka:9092</default_value>
+ </parameter>
+ </parameters>
+ <actions>
+ <action>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <intervalSeconds>0</intervalSeconds>
+ <intervalMinutes>60</intervalMinutes>
+ <DayOfMonth>1</DayOfMonth>
+ <weekDay>1</weekDay>
+ <minutes>0</minutes>
+ <hour>12</hour>
+ <doNotWaitOnFirstExecution>N</doNotWaitOnFirstExecution>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <xloc>128</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+
<filename>${PROJECT_HOME}/prepare-kafka-test-0003-stop-when-idle.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <create_parent_folder>N</create_parent_folder>
+ <set_logfile>N</set_logfile>
+ <set_append_logfile>N</set_append_logfile>
+ <logfile/>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <loglevel>Basic</loglevel>
+ <wait_until_finished>Y</wait_until_finished>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <run_configuration>local</run_configuration>
+ <name>prepare-kafka-test-0003-stop-when-idle.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>336</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <script>
+var txt = previous_result.getLogText();
+
+var ok = false;
+
+var expectedValues = [
+ "Message = Hello Hop Idle!",
+ ];
+
+for (var i = 0 ; i<expectedValues.length ; i++) {
+ var expectedValue = expectedValues[i];
+ if (txt.contains(expectedValue)) {
+ ok = true;
+ log.logBasic("Expected value logged as ''" + expectedValue + "'' FOUND!");
+ }
+}
+
+if (!ok) {
+ log.logBasic("Expected value logged as ''" + expectedValues[0] + "'' NOT
FOUND!");
+}
+
+ok;</script>
+ <name>Check log</name>
+ <description/>
+ <type>EVAL</type>
+ <attributes/>
+ <xloc>864</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+
<filename>${PROJECT_HOME}/0003-kafka-consumer-read-stop-when-idle.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <create_parent_folder>N</create_parent_folder>
+ <set_logfile>N</set_logfile>
+ <set_append_logfile>N</set_append_logfile>
+ <logfile/>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <loglevel>Basic</loglevel>
+ <wait_until_finished>Y</wait_until_finished>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <run_configuration>local</run_configuration>
+ <name>0003-kafka-consumer-read-stop-when-idle.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>608</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>prepare-kafka-test-0003-stop-when-idle.hpl</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>0003-kafka-consumer-read-stop-when-idle.hpl</from>
+ <to>Check log</to>
+ <evaluation>N</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>prepare-kafka-test-0003-stop-when-idle.hpl</from>
+ <to>0003-kafka-consumer-read-stop-when-idle.hpl</to>
+ <evaluation>N</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ </hops>
+ <notepads/>
+ <attributes/>
+</workflow>
diff --git a/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
b/integration-tests/kafka/prepare-kafka-test-0003-stop-when-idle.hpl
similarity index 55%
copy from integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
copy to integration-tests/kafka/prepare-kafka-test-0003-stop-when-idle.hpl
index f10879691b..2b8c534ee6 100644
--- a/integration-tests/kafka/0001-kafka-consumer-read-record-basic.hpl
+++ b/integration-tests/kafka/prepare-kafka-test-0003-stop-when-idle.hpl
@@ -19,22 +19,21 @@ limitations under the License.
-->
<pipeline>
<info>
- <name>0001-kafka-consumer-read-record-basic</name>
+ <name>prepare-kafka-test-0003-stop-when-idle</name>
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
<pipeline_version/>
<pipeline_type>Normal</pipeline_type>
- <pipeline_status>0</pipeline_status>
<parameters>
</parameters>
<capture_transform_performance>N</capture_transform_performance>
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
<created_user>-</created_user>
- <created_date>2021/12/21 09:42:58.199</created_date>
+ <created_date>2021/12/21 09:37:38.673</created_date>
<modified_user>-</modified_user>
- <modified_date>2021/12/21 09:42:58.199</modified_date>
+ <modified_date>2021/12/21 09:37:38.673</modified_date>
<key_for_session_key>H4sIAAAAAAAAAAMAAAAAAAAAAAA=</key_for_session_key>
<is_key_private>N</is_key_private>
</info>
@@ -42,58 +41,14 @@ limitations under the License.
</notepads>
<order>
<hop>
- <from>Kafka Consumer</from>
- <to>Log Output</to>
- <enabled>Y</enabled>
- </hop>
- <hop>
- <from>Log Output</from>
- <to>Stop Kafka Processing</to>
+ <from>Generate rows</from>
+ <to>Kafka Producer</to>
<enabled>Y</enabled>
</hop>
</order>
<transform>
- <name>Kafka Consumer</name>
- <type>KafkaConsumer</type>
- <description/>
- <distribute>Y</distribute>
- <custom_distribution/>
- <copies>1</copies>
- <partitioning>
- <method>none</method>
- <schema_name/>
- </partitioning>
- <topic>hop-test</topic>
- <consumerGroup/>
-
<pipelinePath>${PROJECT_HOME}/0001-kafka-consumer-called-subpipeline.hpl</pipelinePath>
- <subTransform>Out record</subTransform>
- <batchSize>1</batchSize>
- <batchDuration>1000</batchDuration>
- <directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
- <AUTO_COMMIT>N</AUTO_COMMIT>
- <OutputField kafkaName="key" type="String">Key</OutputField>
- <OutputField kafkaName="message" type="String">Message</OutputField>
- <OutputField kafkaName="topic" type="String">Topic</OutputField>
- <OutputField kafkaName="partition" type="Integer">Partition</OutputField>
- <OutputField kafkaName="offset" type="Integer">Offset</OutputField>
- <OutputField kafkaName="timestamp" type="Integer">Timestamp</OutputField>
- <advancedConfig>
- <option property="auto.offset.reset" value="earliest"/>
- <option property="ssl.key.password" value=""/>
- <option property="ssl.keystore.location" value=""/>
- <option property="ssl.keystore.password" value=""/>
- <option property="ssl.truststore.location" value=""/>
- <option property="ssl.truststore.password" value=""/>
- </advancedConfig>
- <attributes/>
- <GUI>
- <xloc>448</xloc>
- <yloc>112</yloc>
- </GUI>
- </transform>
- <transform>
- <name>Log Output</name>
- <type>WriteToLog</type>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
<description/>
<distribute>Y</distribute>
<custom_distribution/>
@@ -102,25 +57,30 @@ limitations under the License.
<method>none</method>
<schema_name/>
</partitioning>
- <loglevel>log_level_basic</loglevel>
- <displayHeader>Y</displayHeader>
- <limitRows>N</limitRows>
- <limitRowsNumber>0</limitRowsNumber>
- <logmessage>Kafka Output</logmessage>
<fields>
<field>
- <name>message</name>
+ <length>-1</length>
+ <name>field</name>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <type>String</type>
+ <nullif>Hello Hop Idle!</nullif>
</field>
</fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <last_time_field>FiveSecondsAgo</last_time_field>
+ <never_ending>N</never_ending>
+ <limit>3</limit>
+ <row_time_field>now</row_time_field>
<attributes/>
<GUI>
- <xloc>560</xloc>
- <yloc>112</yloc>
+ <xloc>288</xloc>
+ <yloc>160</yloc>
</GUI>
</transform>
<transform>
- <name>Stop Kafka Processing</name>
- <type>Abort</type>
+ <name>Kafka Producer</name>
+ <type>KafkaProducerOutput</type>
<description/>
<distribute>Y</distribute>
<custom_distribution/>
@@ -129,14 +89,23 @@ limitations under the License.
<method>none</method>
<schema_name/>
</partitioning>
- <abort_option>SAFE_STOP</abort_option>
- <always_log_rows>Y</always_log_rows>
- <message>Kafka consumer stopped</message>
- <row_threshold>0</row_threshold>
+ <directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
+ <topic>hop-test-idle</topic>
+ <clientId/>
+ <keyField/>
+ <messageField>field</messageField>
+ <advancedConfig>
+ <option property="compression.type" value="none"/>
+ <option property="ssl.key.password" value=""/>
+ <option property="ssl.keystore.location" value=""/>
+ <option property="ssl.keystore.password" value=""/>
+ <option property="ssl.truststore.location" value=""/>
+ <option property="ssl.truststore.password" value=""/>
+ </advancedConfig>
<attributes/>
<GUI>
- <xloc>672</xloc>
- <yloc>112</yloc>
+ <xloc>464</xloc>
+ <yloc>160</yloc>
</GUI>
</transform>
<transform_error_handling>
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
index 95bcd2f40c..7a49d84502 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInput.java
@@ -85,6 +85,9 @@ public class KafkaConsumerInput
data.incomingRowsBuffer = new ArrayList<>();
data.batchDuration = Const.toInt(resolve(meta.getBatchDuration()), 0);
data.batchSize = Const.toInt(resolve(meta.getBatchSize()), 0);
+ data.stopWhenIdle = meta.isStopWhenIdle();
+ data.maxIdleTimeMs = Const.toLong(resolve(meta.getMaxIdleTimeMs()), 500L);
+ data.lastRecordTime = System.currentTimeMillis();
data.consumer = buildKafkaConsumer(this, meta);
@@ -279,16 +282,36 @@ public class KafkaConsumerInput
// Poll records...
// If we get any, process them...
+ // When stop-when-idle is enabled, use a short poll timeout so idle time
can be measured.
//
try {
- Duration duration =
- Duration.ofMillis(data.batchDuration > 0 ? data.batchDuration :
Long.MAX_VALUE);
+ long pollMs =
+ data.stopWhenIdle ? 100L : (data.batchDuration > 0 ?
data.batchDuration : Long.MAX_VALUE);
+ Duration duration = Duration.ofMillis(pollMs);
ConsumerRecords<Object, Object> records = data.consumer.poll(duration);
if (!data.isKafkaConsumerClosing) {
if (records.isEmpty()) {
- // We can just skip this one, poll again next iteration of this
method
+ // No records: optionally stop after max idle time.
+ // Do not count idle until partitions are assigned — group join /
rebalance can take
+ // longer than maxIdleTimeMs and would otherwise stop before any
poll can succeed.
//
+ if (data.stopWhenIdle) {
+ if (data.consumer.assignment() == null ||
data.consumer.assignment().isEmpty()) {
+ data.lastRecordTime = System.currentTimeMillis();
+ } else if ((System.currentTimeMillis() - data.lastRecordTime) >=
data.maxIdleTimeMs) {
+ logBasic(
+ "Kafka consumer idle timeout of "
+ + data.maxIdleTimeMs
+ + "ms exceeded, stopping gracefully");
+ data.isKafkaConsumerClosing = true;
+ if (data.executor != null) {
+ data.executor.getPipeline().stopAll();
+ }
+ setOutputDone();
+ return false;
+ }
+ }
} else {
// Grab the records...
//
@@ -300,6 +323,7 @@ public class KafkaConsumerInput
}
incrementLinesInput();
}
+ data.lastRecordTime = System.currentTimeMillis();
if (isBasic()) {
logBasic("Number of rows read: " +
data.rowProducer.getRowSet().size());
}
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputData.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputData.java
index 85cd933b92..f96ea2b88b 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputData.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputData.java
@@ -31,6 +31,9 @@ public class KafkaConsumerInputData extends BaseTransformData
implements ITransf
public Consumer<Object, Object> consumer;
public int batchDuration;
public int batchSize;
+ public boolean stopWhenIdle;
+ public long maxIdleTimeMs;
+ public long lastRecordTime;
public RowProducer rowProducer;
public SingleThreadedPipelineExecutor executor;
public boolean isKafkaConsumerClosing;
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
index 2c63c7a308..1a5c7e67a2 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
@@ -106,6 +106,9 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
protected TextVar wBatchSize;
protected Label wlBatchDuration;
protected TextVar wBatchDuration;
+ protected Button wStopWhenIdle;
+ protected Label wlMaxIdleTimeMs;
+ protected TextVar wMaxIdleTimeMs;
protected CTabFolder wTabFolder;
protected CTabItem wSetupTab;
@@ -299,6 +302,8 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
m.setExecutionDataProfile(wProfile.getText());
m.setBatchSize(wBatchSize.getText());
m.setBatchDuration(wBatchDuration.getText());
+ m.setStopWhenIdle(wStopWhenIdle.getSelection());
+ m.setMaxIdleTimeMs(wMaxIdleTimeMs.getText());
m.setSubTransform(wSubTransform.getText());
setTopicsFromTable();
@@ -329,7 +334,7 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
wOffsetGroup.setLayout(flOffsetGroup);
FormData fdOffsetGroup = new FormData();
- fdOffsetGroup.top = new FormAttachment(wBatchSize, 15);
+ fdOffsetGroup.top = new FormAttachment(wMaxIdleTimeMs, 15);
fdOffsetGroup.left = new FormAttachment(0, 0);
fdOffsetGroup.right = new FormAttachment(100, 0);
wOffsetGroup.setLayoutData(fdOffsetGroup);
@@ -614,6 +619,34 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
fdBatchSize.top = new FormAttachment(wlBatchSize, 0, SWT.CENTER);
wBatchSize.setLayoutData(fdBatchSize);
+ wStopWhenIdle = new Button(wBatchComp, SWT.CHECK);
+ PropsUi.setLook(wStopWhenIdle);
+ wStopWhenIdle.setText(BaseMessages.getString(PKG,
"KafkaConsumerInputDialog.StopWhenIdle"));
+ wStopWhenIdle.addListener(SWT.Selection, e -> meta.setChanged());
+ FormData fdStopWhenIdle = new FormData();
+ fdStopWhenIdle.left = new FormAttachment(middle, 0);
+ fdStopWhenIdle.top = new FormAttachment(wBatchSize, margin);
+ fdStopWhenIdle.right = new FormAttachment(100, 0);
+ wStopWhenIdle.setLayoutData(fdStopWhenIdle);
+
+ wlMaxIdleTimeMs = new Label(wBatchComp, SWT.RIGHT);
+ PropsUi.setLook(wlMaxIdleTimeMs);
+ wlMaxIdleTimeMs.setText(BaseMessages.getString(PKG,
"KafkaConsumerInputDialog.MaxIdleTimeMs"));
+ FormData fdlMaxIdleTimeMs = new FormData();
+ fdlMaxIdleTimeMs.left = new FormAttachment(0, 0);
+ fdlMaxIdleTimeMs.top = new FormAttachment(wStopWhenIdle, margin);
+ fdlMaxIdleTimeMs.right = new FormAttachment(middle, -margin);
+ wlMaxIdleTimeMs.setLayoutData(fdlMaxIdleTimeMs);
+
+ wMaxIdleTimeMs = new TextVar(variables, wBatchComp, SWT.SINGLE | SWT.LEFT
| SWT.BORDER);
+ PropsUi.setLook(wMaxIdleTimeMs);
+ wMaxIdleTimeMs.addModifyListener(lsMod);
+ FormData fdMaxIdleTimeMs = new FormData();
+ fdMaxIdleTimeMs.left = new FormAttachment(wlMaxIdleTimeMs, margin);
+ fdMaxIdleTimeMs.right = new FormAttachment(100, 0);
+ fdMaxIdleTimeMs.top = new FormAttachment(wlMaxIdleTimeMs, 0, SWT.CENTER);
+ wMaxIdleTimeMs.setLayoutData(fdMaxIdleTimeMs);
+
wBatchComp.layout();
wBatchTab.setControl(wBatchComp);
}
@@ -827,6 +860,8 @@ public class KafkaConsumerInputDialog extends
BaseTransformDialog {
wConsumerGroup.setText(Const.NVL(meta.getConsumerGroup(), ""));
wBatchSize.setText(Const.NVL(meta.getBatchSize(), ""));
wBatchDuration.setText(Const.NVL(meta.getBatchDuration(), ""));
+ wStopWhenIdle.setSelection(meta.isStopWhenIdle());
+ wMaxIdleTimeMs.setText(Const.NVL(meta.getMaxIdleTimeMs(), "500"));
wbAutoCommit.setSelection(meta.isAutoCommit());
wbManualCommit.setSelection(!meta.isAutoCommit());
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
index 6004f74d56..6ee54fe7e3 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMeta.java
@@ -167,6 +167,18 @@ public class KafkaConsumerInputMeta
injectionKeyDescription = "KafkaConsumerInputMeta.Injection.AUTO_COMMIT")
private boolean autoCommit = true;
+ @HopMetadataProperty(
+ key = "stopWhenIdle",
+ injectionKey = "STOP_WHEN_IDLE",
+ injectionKeyDescription =
"KafkaConsumerInputMeta.Injection.STOP_WHEN_IDLE")
+ private boolean stopWhenIdle = false;
+
+ @HopMetadataProperty(
+ key = "maxIdleTimeMs",
+ injectionKey = "MAX_IDLE_TIME_MS",
+ injectionKeyDescription =
"KafkaConsumerInputMeta.Injection.MAX_IDLE_TIME_MS")
+ private String maxIdleTimeMs;
+
@HopMetadataProperty(
groupKey = "options",
key = "option",
@@ -194,6 +206,7 @@ public class KafkaConsumerInputMeta
executionDataProfile = "";
batchSize = "1000";
batchDuration = "1000";
+ maxIdleTimeMs = "500";
subTransform = "";
topics = new ArrayList<>();
options = new ArrayList<>();
@@ -243,6 +256,8 @@ public class KafkaConsumerInputMeta
this.topics = new ArrayList<>(m.topics);
this.consumerGroup = m.consumerGroup;
this.autoCommit = m.autoCommit;
+ this.stopWhenIdle = m.stopWhenIdle;
+ this.maxIdleTimeMs = m.maxIdleTimeMs;
this.mappingMetaRetriever = m.mappingMetaRetriever;
this.options = new ArrayList<>();
m.options.forEach(o -> this.options.add(new KafkaOption(o)));
@@ -327,8 +342,13 @@ public class KafkaConsumerInputMeta
// Check if we get called from error path and only in that case, show
fields that will dump
// the
// record coming from the kafka queue.
- TransformErrorMeta transformErrorMeta =
getParentTransformMeta().getTransformErrorMeta();
+ TransformErrorMeta transformErrorMeta =
+ getParentTransformMeta() != null
+ ? getParentTransformMeta().getTransformErrorMeta()
+ : null;
if (transformErrorMeta != null
+ && transformErrorMeta.getTargetTransform() != null
+ && nextTransform != null
&&
transformErrorMeta.getTargetTransform().getName().equals(nextTransform.getName()))
{
rowMeta.addValueMeta(createValueMetaString(getKeyField().getOutputName()));
rowMeta.addValueMeta(createValueMetaString(getMessageField().getOutputName()));
@@ -398,6 +418,19 @@ public class KafkaConsumerInputMeta
BaseMessages.getString(PKG,
"KafkaConsumerInputMeta.CheckResult.NoBatchDefined"),
transformMeta));
}
+
+ if (isStopWhenIdle()) {
+ try {
+ Long.parseLong(variables.resolve(getMaxIdleTimeMs()));
+ } catch (NumberFormatException e) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG, "KafkaConsumerInputMeta.CheckResult.NaN", "Max idle
time"),
+ transformMeta));
+ }
+ }
}
@Override
diff --git
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
index 7b3fa45c3e..a2eb21bdcf 100644
---
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
+++
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
@@ -24,6 +24,8 @@ KafkaConsumerInputDialog.BatchDuration=Duration (ms)
KafkaConsumerInputDialog.BatchSize=Number of records
KafkaConsumerInputDialog.BatchTab=Batch
KafkaConsumerInputDialog.BootstrapServers=Bootstrap servers
+KafkaConsumerInputDialog.MaxIdleTimeMs=Max idle time (ms)
+KafkaConsumerInputDialog.StopWhenIdle=Stop when idle
KafkaConsumerInputDialog.Column.Name=Output name
KafkaConsumerInputDialog.Column.Ref=Input name
KafkaConsumerInputDialog.Column.Type=Type
@@ -62,6 +64,8 @@ KafkaConsumerInputMeta.Injection.CONNECTION_TYPE=Specify the
connection type: DI
KafkaConsumerInputMeta.Injection.CONSUMER_GROUP=The name of the Kafka consumer
group. This is a required field.
KafkaConsumerInputMeta.Injection.DIRECT_BOOTSTRAP_SERVERS=Specify the
Bootstrap Servers when Connection Type is DIRECT.
KafkaConsumerInputMeta.Injection.BATCH_DURATION=The amount of time to batch
before consuming the messages.
+KafkaConsumerInputMeta.Injection.STOP_WHEN_IDLE=Stop the Kafka consumer when
no records are received for the max idle time.
+KafkaConsumerInputMeta.Injection.MAX_IDLE_TIME_MS=The maximum idle time in
milliseconds before stopping when stop when idle is enabled.
KafkaConsumerInputMeta.Injection.KEY.OUTPUT_NAME=The name of the output field
for the key.
KafkaConsumerInputMeta.Injection.KEY.TYPE=Specify the data type for the key:
String, Integer, Binary, or Number.
KafkaConsumerInputMeta.Injection.MESSAGE.OUTPUT_NAME=The name of the output
field for the message.
diff --git
a/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMetaTest.java
b/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMetaTest.java
index 8a095cc35a..4ab6a3d853 100644
---
a/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMetaTest.java
+++
b/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputMetaTest.java
@@ -104,6 +104,8 @@ class KafkaConsumerInputMetaTest {
meta.setBatchSize("111");
meta.setBatchDuration("222");
meta.setAutoCommit(true);
+ meta.setStopWhenIdle(true);
+ meta.setMaxIdleTimeMs("1500");
meta.getOptions().clear();
meta.getOptions().add(new KafkaOption("auto.offset.reset", "latest"));
meta.getOptions().add(new KafkaOption("ssl.key.password", ""));
@@ -143,6 +145,8 @@ class KafkaConsumerInputMetaTest {
assertEquals("profile", copy.getExecutionDataProfile());
assertEquals("111", copy.getBatchSize());
assertEquals("222", copy.getBatchDuration());
+ assertTrue(copy.isStopWhenIdle());
+ assertEquals("1500", copy.getMaxIdleTimeMs());
assertTrue(copy.getTopics().contains("topic1"));
assertTrue(copy.getTopics().contains("topic2"));
@@ -193,6 +197,9 @@ class KafkaConsumerInputMetaTest {
assertEquals("10000", meta.getBatchDuration());
assertEquals("${KAFKA_SERVER}", meta.getDirectBootstrapServers());
assertFalse(meta.isAutoCommit());
+ // Legacy XML without the new fields keeps defaults
+ assertFalse(meta.isStopWhenIdle());
+ assertEquals("500", meta.getMaxIdleTimeMs());
assertEquals(6, meta.getOptions().size());
assertEquals("auto.offset.reset",
meta.getOptions().getFirst().getProperty());
@@ -228,4 +235,22 @@ class KafkaConsumerInputMetaTest {
assertEquals(KafkaConsumerField.Type.Integer, f.getOutputType());
assertEquals(KafkaConsumerField.Name.TIMESTAMP, f.getKafkaName());
}
+
+ @Test
+ void testStopWhenIdleDefaults() {
+ KafkaConsumerInputMeta meta = new KafkaConsumerInputMeta();
+ assertFalse(meta.isStopWhenIdle());
+ assertEquals("500", meta.getMaxIdleTimeMs());
+ }
+
+ @Test
+ void testStopWhenIdleClone() {
+ KafkaConsumerInputMeta meta = new KafkaConsumerInputMeta();
+ meta.setStopWhenIdle(true);
+ meta.setMaxIdleTimeMs("2500");
+
+ KafkaConsumerInputMeta copy = meta.clone();
+ assertTrue(copy.isStopWhenIdle());
+ assertEquals("2500", copy.getMaxIdleTimeMs());
+ }
}
