This is an automated email from the ASF dual-hosted git repository.
mattcasters 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 198bccdf82 Issue #2679 : Read the Kafka producer topic from a field
(#7747)
198bccdf82 is described below
commit 198bccdf8232e382eff7f11e7b6d5a54ec9fbc30
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Sat Aug 1 16:00:09 2026 -0400
Issue #2679 : Read the Kafka producer topic from a field (#7747)
* Issue #2679 : Read the Kafka producer topic from a field
The Kafka producer publishes every row to one configured topic, so routing
rows
to different topics means one producer transform per topic.
Add a "Get topic from field" option. When it is enabled the topic is read
from a
field on each incoming row, letting a single transform fan rows out across
topics; when it is disabled the transform behaves exactly as before. The
option
follows the naming already used for the same idea in Table Output, which
pairs a
tableNameInField flag with a tableNameField name.
The topic is only ever part of the ProducerRecord, never of the producer
configuration, so a single producer instance serves every topic and no extra
connections are created.
Field resolution and validation happen once, when the first row arrives: an
enabled option with no field configured, or naming a field that is not in
the
input stream, fails the transform with a clear message rather than at send
time.
A row whose topic field is empty fails with the field name and the line
number.
The dialog disables whichever of the two topic widgets does not apply and
validates the field against the incoming stream before accepting the change.
Existing pipelines carry no topicInField element and so keep using the fixed
topic name.
* Issue #2679 : Add an integration test for the Kafka topic-from-field
option
Publishes two rows through a single Kafka Producer, each naming its own
topic in
a field, then consumes only the first topic.
The check asserts that topic A's message arrives and that topic B's does
not.
Asserting only the first would also pass if the producer ignored the field
and
sent every row to one topic, which is the regression worth catching. The
static
topic is set to a name nothing reads, so a row that skipped the field lands
somewhere the test cannot see and the assertion fails.
Reuses 0001-kafka-consumer-called-subpipeline.hpl, which is a plain injector
over the standard Kafka output fields and needs nothing test specific.
---
.../pages/pipeline/transforms/kafkaproducer.adoc | 11 +-
.../kafka/0004-kafka-consumer-topic-from-field.hpl | 123 +++++++++++++++
.../main-0004-kafka-test-topic-from-field.hwf | 174 +++++++++++++++++++++
.../prepare-kafka-test-0004-topic-from-field.hpl | 127 +++++++++++++++
.../kafka/producer/KafkaProducerOutput.java | 51 +++++-
.../kafka/producer/KafkaProducerOutputData.java | 2 +
.../kafka/producer/KafkaProducerOutputDialog.java | 96 +++++++++++-
.../kafka/producer/KafkaProducerOutputMeta.java | 14 ++
.../producer/messages/messages_en_US.properties | 9 ++
.../producer/KafkaProducerOutputMetaTest.java | 29 ++++
.../src/test/resources/kafka-producer-output.xml | 2 +
11 files changed, 632 insertions(+), 6 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
index e83994a1e1..fdb99fa38a 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/kafkaproducer.adoc
@@ -27,7 +27,7 @@ under the License.
The Kafka producer transform allows you to publish messages in near-real-time
across worker nodes where multiple, subscribed members have access.
-A Kafka producer transform publishes a stream of records to one Kafka topic.
+A Kafka producer transform publishes a stream of records to one Kafka topic,
or to a topic read per row from a field in the incoming stream.
|
== Supported Engines
@@ -59,6 +59,15 @@ A Kafka producer transform publishes a stream of records to
one Kafka topic.
|Topic
|The category to which records are published.
+Disabled when the topic is read from a field.
+
+|Get topic from field
+|Read the topic from a field in the incoming stream instead of publishing
every row to a fixed topic.
+This lets a single transform route rows across topics, rather than needing one
Kafka producer transform per topic.
+
+|Topic field
+|The field holding the topic name for each row, used when *Get topic from
field* is enabled.
+Rows where this field is empty cause the transform to fail.
|Key Field
|In Kafka, all messages can be keyed, allowing for messages to be distributed
to partitions based on their keys in a default routing scheme.
diff --git a/integration-tests/kafka/0004-kafka-consumer-topic-from-field.hpl
b/integration-tests/kafka/0004-kafka-consumer-topic-from-field.hpl
new file mode 100644
index 0000000000..3d1c48bbc0
--- /dev/null
+++ b/integration-tests/kafka/0004-kafka-consumer-topic-from-field.hpl
@@ -0,0 +1,123 @@
+<?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>0004-kafka-consumer-topic-from-field</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Reads only topic A, so the log proves the producer routed per
row rather than publishing everything to one topic.</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>2026/08/01 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/01 12:00:00.000</modified_date>
+ <key_for_session_key>H4sIAAAAAAAAAAMAAAAAAAAAAAA=</key_for_session_key>
+ <is_key_private>N</is_key_private>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Kafka Consumer</from>
+ <to>Log Output</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-a</topic>
+ <consumerGroup/>
+
<pipelinePath>${PROJECT_HOME}/0001-kafka-consumer-called-subpipeline.hpl</pipelinePath>
+ <subTransform>Out record</subTransform>
+ <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>
+ <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>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <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>
+ </field>
+ <field>
+ <name>topic</name>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>560</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/kafka/main-0004-kafka-test-topic-from-field.hwf
b/integration-tests/kafka/main-0004-kafka-test-topic-from-field.hwf
new file mode 100644
index 0000000000..3371be4510
--- /dev/null
+++ b/integration-tests/kafka/main-0004-kafka-test-topic-from-field.hwf
@@ -0,0 +1,174 @@
+<?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-0004-kafka-test-topic-from-field</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Checks that the Kafka Producer routes each row to the topic
named in a field.</description>
+ <extended_description/>
+ <created_user>-</created_user>
+ <modified_user>-</modified_user>
+ <created_date>2026/08/01 12:00:00.000</created_date>
+ <modified_date>2026/08/01 12:00:00.000</modified_date>
+ <workflow_version/>
+ <parameters>
+ <parameter>
+ <name>BOOTSTRAP_SERVERS</name>
+ <description>Kafka bootstrap servers (docker IT network
hostname)</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-0004-topic-from-field.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-0004-topic-from-field.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>336</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+
<filename>${PROJECT_HOME}/0004-kafka-consumer-topic-from-field.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>0004-kafka-consumer-topic-from-field.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>608</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <script>
+var txt = previous_result.getLogText();
+
+// Only topic A was consumed. Seeing A's message proves the row was routed by
its
+// field; not seeing B's proves the other row went somewhere else rather than
+// every row landing on one topic.
+var expectedValue = "Message = Hello topic A!";
+var unexpectedValue = "Message = Hello topic B!";
+
+var routedToTopicA = txt.contains(expectedValue);
+var leakedTopicB = txt.contains(unexpectedValue);
+
+if (routedToTopicA) {
+ log.logBasic("Expected value logged as ''" + expectedValue + "'' FOUND!");
+} else {
+ log.logBasic("Expected value logged as ''" + expectedValue + "'' NOT
FOUND!");
+}
+
+if (leakedTopicB) {
+ log.logBasic("Unexpected value logged as ''" + unexpectedValue + "'' FOUND
on topic A!");
+}
+
+var ok = routedToTopicA && !leakedTopicB;
+
+ok;</script>
+ <name>Check log</name>
+ <description/>
+ <type>EVAL</type>
+ <attributes/>
+ <xloc>864</xloc>
+ <yloc>144</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>prepare-kafka-test-0004-topic-from-field.hpl</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>prepare-kafka-test-0004-topic-from-field.hpl</from>
+ <to>0004-kafka-consumer-topic-from-field.hpl</to>
+ <evaluation>N</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>0004-kafka-consumer-topic-from-field.hpl</from>
+ <to>Check log</to>
+ <evaluation>N</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ </hops>
+ <notepads/>
+ <attributes/>
+</workflow>
diff --git
a/integration-tests/kafka/prepare-kafka-test-0004-topic-from-field.hpl
b/integration-tests/kafka/prepare-kafka-test-0004-topic-from-field.hpl
new file mode 100644
index 0000000000..e6ac77a72f
--- /dev/null
+++ b/integration-tests/kafka/prepare-kafka-test-0004-topic-from-field.hpl
@@ -0,0 +1,127 @@
+<?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>prepare-kafka-test-0004-topic-from-field</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Publishes two rows through a single Kafka Producer, each
carrying its own target topic in a field.</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>2026/08/01 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/08/01 12:00:00.000</modified_date>
+ <key_for_session_key>H4sIAAAAAAAAAAMAAAAAAAAAAAA=</key_for_session_key>
+ <is_key_private>N</is_key_private>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Rows with topics</from>
+ <to>Kafka Producer</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Rows with topics</name>
+ <type>DataGrid</type>
+ <description>Two messages destined for two different topics</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>hop-test-topic-a</item>
+ <item>Hello topic A!</item>
+ </line>
+ <line>
+ <item>hop-test-topic-b</item>
+ <item>Hello topic B!</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>target_topic</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>field</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>288</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Kafka Producer</name>
+ <type>KafkaProducerOutput</type>
+ <description>Routes each row to the topic named in the target_topic
field</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
+ <topic>hop-test-unused-fallback</topic>
+ <topicInField>Y</topicInField>
+ <topicField>target_topic</topicField>
+ <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>464</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
index 26b9e6820f..b76e2a9301 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutput.java
@@ -67,6 +67,23 @@ public class KafkaProducerOutput
data.keyValueMeta = getInputRowMeta().getValueMeta(data.keyFieldIndex);
data.msgValueMeta =
getInputRowMeta().getValueMeta(data.messageFieldIndex);
+ if (meta.isTopicInField()) {
+ String topicField = resolve(meta.getTopicField());
+ if (StringUtils.isEmpty(topicField)) {
+ throw new HopException(
+ BaseMessages.getString(PKG,
"KafkaProducerOutput.Error.NoTopicFieldSpecified"));
+ }
+ data.topicFieldIndex = getInputRowMeta().indexOfValue(topicField);
+ if (data.topicFieldIndex < 0) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "KafkaProducerOutput.Error.TopicFieldNotFound",
topicField));
+ }
+ } else {
+ data.topicFieldIndex = -1;
+ data.topic = resolve(meta.getTopic());
+ }
+
data.kafkaProducer =
kafkaFactory.producer(
meta,
@@ -82,12 +99,14 @@ public class KafkaProducerOutput
if (!data.isOpen) {
return false;
}
+ String topic = resolveTopic(r);
+
ProducerRecord<Object, Object> producerRecord;
// allow for null keys
if (data.keyFieldIndex < 0
|| getInputRowMeta().isNull(r, data.keyFieldIndex)
|| StringUtils.isEmpty(r[data.keyFieldIndex].toString())) {
- producerRecord = new ProducerRecord<>(resolve(meta.getTopic()),
r[data.messageFieldIndex]);
+ producerRecord = new ProducerRecord<>(topic, r[data.messageFieldIndex]);
} else {
Object nativeObject =
@@ -97,9 +116,7 @@ public class KafkaProducerOutput
producerRecord =
new ProducerRecord<>(
- resolve(meta.getTopic()),
- getInputRowMeta().getString(r, data.keyFieldIndex),
- nativeObject);
+ topic, getInputRowMeta().getString(r, data.keyFieldIndex),
nativeObject);
}
data.kafkaProducer.send(producerRecord);
@@ -114,6 +131,32 @@ public class KafkaProducerOutput
return true;
}
+ /**
+ * Determines the topic the given row is sent to. With "topic from field"
disabled this is the
+ * configured topic, resolved once when the first row arrives. With it
enabled the topic is read
+ * from the row itself, so a single transform can fan rows out across topics
instead of needing
+ * one Kafka Producer per topic.
+ *
+ * @param r the current input row
+ * @return the topic name for this row
+ * @throws HopException if the row carries no usable topic
+ */
+ private String resolveTopic(Object[] r) throws HopException {
+ if (data.topicFieldIndex < 0) {
+ return data.topic;
+ }
+ String topic = getInputRowMeta().getString(r, data.topicFieldIndex);
+ if (StringUtils.isEmpty(topic)) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG,
+ "KafkaProducerOutput.Error.EmptyTopicInField",
+ getInputRowMeta().getValueMeta(data.topicFieldIndex).getName(),
+ getLinesRead()));
+ }
+ return topic;
+ }
+
@Override
public void stopRunning() {
if (data.kafkaProducer != null && data.isOpen) {
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
index 507f50c938..20bc68c198 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputData.java
@@ -29,6 +29,8 @@ public class KafkaProducerOutputData extends
BaseTransformData implements ITrans
public Producer<Object, Object> kafkaProducer;
public int keyFieldIndex;
public int messageFieldIndex;
+ public int topicFieldIndex;
+ public String topic;
public boolean isOpen;
public KafkaProducerOutputData() {
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
index e8b38409f1..c00a064b2f 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputDialog.java
@@ -49,6 +49,7 @@ import org.eclipse.swt.events.ModifyListener;
import org.eclipse.swt.layout.FormAttachment;
import org.eclipse.swt.layout.FormData;
import org.eclipse.swt.layout.FormLayout;
+import org.eclipse.swt.widgets.Button;
import org.eclipse.swt.widgets.Composite;
import org.eclipse.swt.widgets.Label;
import org.eclipse.swt.widgets.Listener;
@@ -71,6 +72,9 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
private TextVar wClientId;
private ComboVar wTopic;
+ private Button wTopicInField;
+ private Label wlTopicField;
+ private ComboVar wTopicField;
private ComboVar wKeyField;
private ComboVar wMessageField;
private TableView optionsTable;
@@ -203,12 +207,60 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
kafkaDialogHelper.clusterNameChanged(event);
});
+ Label wlTopicInField = new Label(wSetupComp, SWT.RIGHT);
+ PropsUi.setLook(wlTopicInField);
+ wlTopicInField.setText(BaseMessages.getString(PKG,
"KafkaProducerOutputDialog.TopicInField"));
+ FormData fdlTopicInField = new FormData();
+ fdlTopicInField.left = new FormAttachment(0, 0);
+ fdlTopicInField.top = new FormAttachment(wTopic, margin);
+ fdlTopicInField.right = new FormAttachment(middle, -margin);
+ wlTopicInField.setLayoutData(fdlTopicInField);
+
+ wTopicInField = new Button(wSetupComp, SWT.CHECK);
+ PropsUi.setLook(wTopicInField);
+ FormData fdTopicInField = new FormData();
+ fdTopicInField.left = new FormAttachment(wlTopicInField, margin);
+ fdTopicInField.top = new FormAttachment(wlTopicInField, 0, SWT.CENTER);
+ fdTopicInField.right = new FormAttachment(100, 0);
+ wTopicInField.setLayoutData(fdTopicInField);
+ wTopicInField.addListener(
+ SWT.Selection,
+ e -> {
+ meta.setChanged();
+ enableFields();
+ });
+
+ wlTopicField = new Label(wSetupComp, SWT.RIGHT);
+ PropsUi.setLook(wlTopicField);
+ wlTopicField.setText(BaseMessages.getString(PKG,
"KafkaProducerOutputDialog.TopicField"));
+ FormData fdlTopicField = new FormData();
+ fdlTopicField.left = new FormAttachment(0, 0);
+ fdlTopicField.top = new FormAttachment(wTopicInField, margin);
+ fdlTopicField.right = new FormAttachment(middle, -margin);
+ wlTopicField.setLayoutData(fdlTopicField);
+
+ wTopicField = new ComboVar(variables, wSetupComp, SWT.SINGLE | SWT.LEFT |
SWT.BORDER);
+ PropsUi.setLook(wTopicField);
+ wTopicField.addModifyListener(lsMod);
+ FormData fdTopicField = new FormData();
+ fdTopicField.left = new FormAttachment(wlTopicField, margin);
+ fdTopicField.top = new FormAttachment(wlTopicField, 0, SWT.CENTER);
+ fdTopicField.right = new FormAttachment(100, 0);
+ wTopicField.setLayoutData(fdTopicField);
+ wTopicField
+ .getCComboWidget()
+ .addListener(
+ SWT.FocusIn,
+ e ->
+ KafkaDialogHelper.populateFieldsList(
+ variables, pipelineMeta, wTopicField, transformName));
+
Label wlKeyField = new Label(wSetupComp, SWT.RIGHT);
PropsUi.setLook(wlKeyField);
wlKeyField.setText(BaseMessages.getString(PKG,
"KafkaProducerOutputDialog.KeyField"));
FormData fdlKeyField = new FormData();
fdlKeyField.left = new FormAttachment(0, 0);
- fdlKeyField.top = new FormAttachment(wTopic, margin);
+ fdlKeyField.top = new FormAttachment(wTopicField, margin);
fdlKeyField.right = new FormAttachment(middle, -margin);
wlKeyField.setLayoutData(fdlKeyField);
@@ -362,10 +414,24 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
wBootstrapServers.setText(Const.NVL(meta.getDirectBootstrapServers(), ""));
wClientId.setText(Const.NVL(meta.getClientId(), ""));
wTopic.setText(Const.NVL(meta.getTopic(), ""));
+ wTopicInField.setSelection(meta.isTopicInField());
+ wTopicField.setText(Const.NVL(meta.getTopicField(), ""));
wKeyField.setText(Const.NVL(meta.getKeyField(), ""));
wMessageField.setText(Const.NVL(meta.getMessageField(), ""));
populateOptionsData();
+ enableFields();
+ }
+
+ /**
+ * The topic is either a fixed name or read from a stream field, never both,
so only the relevant
+ * widget stays enabled.
+ */
+ private void enableFields() {
+ boolean topicInField = wTopicInField.getSelection();
+ wTopic.setEnabled(!topicInField);
+ wlTopicField.setEnabled(topicInField);
+ wTopicField.setEnabled(topicInField);
}
private boolean checkMissingField(String fieldName) {
@@ -410,6 +476,34 @@ public class KafkaProducerOutputDialog extends
BaseTransformDialog {
meta.setDirectBootstrapServers(wBootstrapServers.getText());
meta.setClientId(wClientId.getText());
meta.setTopic(wTopic.getText());
+
+ if (wTopicInField.getSelection()) {
+ if (Utils.isEmpty(wTopicField.getText())) {
+ MessageBox mb = new MessageBox(shell, SWT.OK | SWT.ICON_ERROR);
+ mb.setMessage(
+ BaseMessages.getString(PKG,
"KafkaProducerOutputDialog.TopicFieldMandatory.Message"));
+ mb.setText(
+ BaseMessages.getString(PKG,
CONST_KAFKA_PRODUCER_OUTPUT_DIALOG_FIELD_NOT_EXISTS_TITLE));
+ mb.open();
+ return;
+ }
+ if (checkMissingField(wTopicField.getText())) {
+ MessageBox mb = new MessageBox(shell, SWT.OK | SWT.ICON_ERROR);
+ mb.setMessage(
+ BaseMessages.getString(
+ PKG,
+ "KafkaProducerOutputDialog.TopicFieldNotExists.Message",
+ wTopicField.getText()));
+ mb.setText(
+ BaseMessages.getString(PKG,
CONST_KAFKA_PRODUCER_OUTPUT_DIALOG_FIELD_NOT_EXISTS_TITLE));
+ mb.open();
+ return;
+ }
+ }
+
+ meta.setTopicInField(wTopicInField.getSelection());
+ meta.setTopicField(wTopicField.getText());
+
if (!Utils.isEmpty(wKeyField.getText()) &&
checkMissingField(wKeyField.getText())) {
MessageBox mb = new MessageBox(shell, SWT.OK | SWT.ICON_ERROR);
mb.setMessage(
diff --git
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
index c37353fd12..625329f7af 100644
---
a/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
+++
b/plugins/transforms/kafka/src/main/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMeta.java
@@ -69,6 +69,18 @@ public class KafkaProducerOutputMeta
injectionKeyDescription = "KafkaProducerOutputMeta.Injection.TOPIC")
private String topic;
+ @HopMetadataProperty(
+ key = "topicInField",
+ injectionKey = "TOPIC_IN_FIELD",
+ injectionKeyDescription =
"KafkaProducerOutputMeta.Injection.TOPIC_IN_FIELD")
+ private boolean topicInField;
+
+ @HopMetadataProperty(
+ key = "topicField",
+ injectionKey = "TOPIC_FIELD",
+ injectionKeyDescription =
"KafkaProducerOutputMeta.Injection.TOPIC_FIELD")
+ private String topicField;
+
@HopMetadataProperty(
key = "keyField",
injectionKey = "KEY_FIELD",
@@ -98,6 +110,8 @@ public class KafkaProducerOutputMeta
this.directBootstrapServers = m.directBootstrapServers;
this.clientId = m.clientId;
this.topic = m.topic;
+ this.topicInField = m.topicInField;
+ this.topicField = m.topicField;
this.keyField = m.keyField;
this.messageField = m.messageField;
m.options.forEach(option -> this.options.add(new KafkaOption(option)));
diff --git
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
index 950a000ffc..7b62ef7716 100644
---
a/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
+++
b/plugins/transforms/kafka/src/main/resources/org/apache/hop/pipeline/transforms/kafka/producer/messages/messages_en_US.properties
@@ -18,6 +18,9 @@
KafkaConsumerOutput.Log.LineNumber=Linenr {0}
KafkaProducer.TypeLongDesc=Kafka producer
KafkaProducer.TypeTooltipDesc=Sends messages to a Kafka producer for a certain
topic
+KafkaProducerOutput.Error.EmptyTopicInField=Field ''{0}'' contains no topic
name for the row at line {1}
+KafkaProducerOutput.Error.NoTopicFieldSpecified=No field containing the topic
name has been specified
+KafkaProducerOutput.Error.TopicFieldNotFound=Field ''{0}'' containing the
topic name is not present in the input stream
KafkaProducerOutputDialog.BootstrapServerMandatory.Message=''Bootstrap
servers'' field''s value is mandatory!
KafkaProducerOutputDialog.BootstrapServers=Bootstrap servers
KafkaProducerOutputDialog.ClientId=Client ID
@@ -35,6 +38,10 @@ KafkaProducerOutputDialog.Options.Tab=Options
KafkaProducerOutputDialog.SetupTab=Setup
KafkaProducerOutputDialog.Shell.Title=Kafka producer
KafkaProducerOutputDialog.Topic=Topic
+KafkaProducerOutputDialog.TopicField=Topic field
+KafkaProducerOutputDialog.TopicFieldMandatory.Message=Field containing the
topic name is mandatory when the topic is read from a field!
+KafkaProducerOutputDialog.TopicFieldNotExists.Message=Field ''{0}'' containing
the topic name is not present in input stream
+KafkaProducerOutputDialog.TopicInField=Get topic from field
KafkaProducerOutputDialog.TransformName.Label=Transform name
KafkaProducerOutputMeta.Injection.CLIENT_ID=The ID used to identify and track
requests made to the server.
KafkaProducerOutputMeta.Injection.CLUSTER_NAME=The cluster containing the
Kafka connections details.
@@ -45,6 +52,8 @@ KafkaProducerOutputMeta.Injection.KEY_FIELD=The name of the
stream field used as
KafkaProducerOutputMeta.Injection.MESSAGE_FIELD=The name of the stream field
used as the Kafka message.
KafkaProducerOutputMeta.Injection.NAMES=The name(s) of the Kafka producer
configuration properties.
KafkaProducerOutputMeta.Injection.TOPIC=The name of the topic to publish to.
+KafkaProducerOutputMeta.Injection.TOPIC_FIELD=The name of the stream field
holding the topic to publish to, used when TOPIC_IN_FIELD is enabled.
+KafkaProducerOutputMeta.Injection.TOPIC_IN_FIELD=Read the topic from a stream
field instead of using a fixed topic name.
KafkaProducerOutputMeta.Injection.VALUES=The value(s) of the Kafka producer
configuration properties.
KafkaConsumerInputMeta.Injection.CONFIGURATION_PROPERTIES=Configuration options
KafkaProducerOutputMeta.keyword=kafka,producer,publish,topic,streaming
diff --git
a/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMetaTest.java
b/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMetaTest.java
index 2c566ce344..85fb49e5cc 100644
---
a/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMetaTest.java
+++
b/plugins/transforms/kafka/src/test/java/org/apache/hop/pipeline/transforms/kafka/producer/KafkaProducerOutputMetaTest.java
@@ -18,6 +18,9 @@
package org.apache.hop.pipeline.transforms.kafka.producer;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
import org.junit.jupiter.api.Test;
@@ -31,6 +34,8 @@ class KafkaProducerOutputMetaTest {
assertEquals("${KAFKA_SERVER}", meta.getDirectBootstrapServers());
assertEquals("${KAFKA_TOPIC}", meta.getTopic());
+ assertTrue(meta.isTopicInField());
+ assertEquals("topic", meta.getTopicField());
assertEquals("Hop", meta.getClientId());
assertEquals("id", meta.getKeyField());
assertEquals("json", meta.getMessageField());
@@ -39,4 +44,28 @@ class KafkaProducerOutputMetaTest {
assertEquals("none", meta.getOptions().getFirst().getValue());
assertEquals("ssl.truststore.password",
meta.getOptions().getLast().getProperty());
}
+
+ @Test
+ void testTopicInFieldDefaultsToOff() {
+ // Pipelines saved before the topic-from-field option existed carry no
topicInField
+ // element, so the default has to keep using the fixed topic name.
+ KafkaProducerOutputMeta meta = new KafkaProducerOutputMeta();
+
+ assertFalse(meta.isTopicInField());
+ assertNull(meta.getTopicField());
+ }
+
+ @Test
+ void testCopyConstructorCopiesTopicFromFieldSettings() {
+ KafkaProducerOutputMeta meta = new KafkaProducerOutputMeta();
+ meta.setTopic("fallback");
+ meta.setTopicInField(true);
+ meta.setTopicField("topic");
+
+ KafkaProducerOutputMeta copy = new KafkaProducerOutputMeta(meta);
+
+ assertEquals("fallback", copy.getTopic());
+ assertTrue(copy.isTopicInField());
+ assertEquals("topic", copy.getTopicField());
+ }
}
diff --git
a/plugins/transforms/kafka/src/test/resources/kafka-producer-output.xml
b/plugins/transforms/kafka/src/test/resources/kafka-producer-output.xml
index b2e38bcd9a..a5b0aa37bc 100644
--- a/plugins/transforms/kafka/src/test/resources/kafka-producer-output.xml
+++ b/plugins/transforms/kafka/src/test/resources/kafka-producer-output.xml
@@ -18,6 +18,8 @@
<transform>
<directBootstrapServers>${KAFKA_SERVER}</directBootstrapServers>
<topic>${KAFKA_TOPIC}</topic>
+ <topicInField>Y</topicInField>
+ <topicField>topic</topicField>
<clientId>Hop</clientId>
<keyField>id</keyField>
<messageField>json</messageField>
