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 3d26193fcd issue #4120 : Simple mapping not receiving output (sub-sub
in Kafka Consumer) (#7379)
3d26193fcd is described below
commit 3d26193fcd1397137ae6dd3aa47832864ced0c44
Author: Matt Casters <[email protected]>
AuthorDate: Fri Jul 3 07:58:43 2026 +0200
issue #4120 : Simple mapping not receiving output (sub-sub in Kafka
Consumer) (#7379)
---
assemblies/plugins/pom.xml | 6 +
.../integration-tests/integration-tests-kafka.yaml | 4 +-
.../pipeline-run-configurations.adoc | 1 +
.../single-threaded-pipeline-engine.adoc | 45 ++++++
.../pipeline/SingleThreadedPipelineExecutor.java | 13 +-
.../IMeasuringLocalPipelineRunConfiguration.java | 25 +++
.../local/LocalPipelineRunConfiguration.java | 167 +--------------------
.../LocalSinglePipelineRunConfiguration.java | 45 ------
...ka-consumer-called-subpipeline-with-mapping.hpl | 154 +++++++++++++++++++
.../0002-kafka-consumer-read-record-mapping.hpl | 166 ++++++++++++++++++++
.../kafka/0002-mapping-subpipeline.hpl | 134 +++++++++++++++++
integration-tests/kafka/disabled.txt | 17 ---
.../kafka/main-0001-kafka-test-basic.hwf | 81 +++++-----
...-basic.hwf => main-0002-kafka-test-mapping.hwf} | 97 ++++++------
.../kafka/prepare-kafka-test-0002.hpl | 113 ++++++++++++++
plugins/engines/pom.xml | 1 +
plugins/engines/{ => single-threaded}/pom.xml | 26 +++-
.../src/assembly/assembly.xml} | 36 +++--
.../SingleThreadedPipelineEngine.java | 46 +++---
.../SingleThreadedPipelineRunConfiguration.java | 94 ++++++++++++
.../InjectDataSetIntoTransformExtensionPoint.java | 10 +-
.../pipeline/transforms/mapping/SimpleMapping.java | 20 ++-
.../file/shared/PipelineRowSamplerHelper.java | 10 +-
23 files changed, 924 insertions(+), 387 deletions(-)
diff --git a/assemblies/plugins/pom.xml b/assemblies/plugins/pom.xml
index ee02975c13..3943d9fdf0 100644
--- a/assemblies/plugins/pom.xml
+++ b/assemblies/plugins/pom.xml
@@ -606,6 +606,12 @@
<version>${project.version}</version>
<type>zip</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-engines-single-threaded</artifactId>
+ <version>${project.version}</version>
+ <type>zip</type>
+ </dependency>
<dependency>
<groupId>org.apache.hop</groupId>
<artifactId>hop-misc-async</artifactId>
diff --git a/docker/integration-tests/integration-tests-kafka.yaml
b/docker/integration-tests/integration-tests-kafka.yaml
index e0327a3589..0273ec87cb 100644
--- a/docker/integration-tests/integration-tests-kafka.yaml
+++ b/docker/integration-tests/integration-tests-kafka.yaml
@@ -30,8 +30,8 @@ services:
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
- KAFKA_LISTENERS: PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
- KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+ KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:
CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/pipeline-run-configurations.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/pipeline-run-configurations.adoc
index efa1f8dd57..a92a49d52f 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/pipeline-run-configurations.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/pipeline-run-configurations.adoc
@@ -58,6 +58,7 @@ The available engine types are:
*
*xref:pipeline/pipeline-run-configurations/beam-spark-pipeline-engine.adoc[Beam
Spark pipeline engine]*: this configuration runs pipelines on Apache Spark over
Apache Beam
*
*xref:pipeline/pipeline-run-configurations/native-local-pipeline-engine.adoc[Hop
local pipeline engine]*: this configuration runs pipelines locally in the
native Hop engine
*
*xref:pipeline/pipeline-run-configurations/native-remote-pipeline-engine.adoc[Hop
remote pipeline engine]*: this configuration runs pipelines in the native Hop
engine on a remote machine
+*
*xref:pipeline/pipeline-run-configurations/single-threaded-pipeline-engine.adoc[Hop
local single threaded pipeline engine]*: this configuration runs pipelines
locally in a single-threaded fashion
|===
TIP: Check the
link:https://beam.apache.org/documentation/runners/capability-matrix/[Beam
Capability Matrix] to help you decide which Beam engine works best for your
pipeline.
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/single-threaded-pipeline-engine.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/single-threaded-pipeline-engine.adoc
new file mode 100644
index 0000000000..af52e9361e
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/single-threaded-pipeline-engine.adoc
@@ -0,0 +1,45 @@
+////
+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.
+////
+[[SingleThreadedPipelineEngine]]
+:imagesdir: ../assets/images
+:description: The single-threaded engine runs Hop pipelines locally in a
single thread, executing transforms sequentially in topological order.
+
+= Single Threaded Pipeline Engine
+
+== Single Threaded
+
+The single-threaded pipeline engine runs your pipeline locally in a
single-threaded fashion.
+Every transform is executed in topological order, and data is processed
sequentially rather than in a parallel stream.
+
+This execution mode is useful for debugging, testing, or processing batch data
where concurrency overhead is undesirable, as well as executing pipelines
embedded in other environments (such as mapping transforms in Kafka message
processing).
+
+=== Options
+
+|===
+|Option|Description|Default
+
+|Sample type while running in the GUI
+|Here you can specify which rows you want to see when you click on the little
grid icons on the transform icons during a pipeline execution.
+You can choose: `None`: do not show, `First` : show the first rows, `Last` :
show the last rows or `Random`: a random selection of rows (reservoir sampling)
+|Last
+
+|Number of rows to sample in the GUI
+|The number of rows which will be collected
+|100
+
+|===
+
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/SingleThreadedPipelineExecutor.java
b/engine/src/main/java/org/apache/hop/pipeline/SingleThreadedPipelineExecutor.java
index 5b63914841..e0b0d51259 100644
---
a/engine/src/main/java/org/apache/hop/pipeline/SingleThreadedPipelineExecutor.java
+++
b/engine/src/main/java/org/apache/hop/pipeline/SingleThreadedPipelineExecutor.java
@@ -30,6 +30,8 @@ import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMetaDataCombi;
import org.apache.hop.pipeline.transform.stream.IStream;
+@Getter
+@Setter
public class SingleThreadedPipelineExecutor {
private List<TransformMetaDataCombi> transforms;
@@ -377,7 +379,12 @@ public class SingleThreadedPipelineExecutor {
// Signal the transform that a batch of rows has passed for this
iteration (sort rows and
// all)
//
- combi.transform.batchComplete();
+ try {
+ combi.transform.batchComplete();
+ } catch (Exception e) {
+ combi.transform.setErrors(1);
+ throw e;
+ }
if (transformDone) {
nrDone++;
@@ -495,10 +502,6 @@ public class SingleThreadedPipelineExecutor {
}
}
- public Pipeline getPipeline() {
- return pipeline;
- }
-
/** Clear the error in the pipeline, clear all the rows from all the row
sets... */
public void clearError() {
pipeline.clearError();
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engines/IMeasuringLocalPipelineRunConfiguration.java
b/engine/src/main/java/org/apache/hop/pipeline/engines/IMeasuringLocalPipelineRunConfiguration.java
new file mode 100644
index 0000000000..fa9ce7ca1e
--- /dev/null
+++
b/engine/src/main/java/org/apache/hop/pipeline/engines/IMeasuringLocalPipelineRunConfiguration.java
@@ -0,0 +1,25 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.hop.pipeline.engines;
+
+public interface IMeasuringLocalPipelineRunConfiguration {
+ String getSampleTypeInGui();
+
+ String getSampleSize();
+}
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
index e87b89d90b..7b1ef1b413 100644
---
a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
+++
b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
@@ -19,6 +19,8 @@ package org.apache.hop.pipeline.engines.local;
import java.util.ArrayList;
import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
import org.apache.hop.core.Const;
import org.apache.hop.core.gui.plugin.GuiElementType;
import org.apache.hop.core.gui.plugin.GuiPlugin;
@@ -32,8 +34,11 @@ import
org.apache.hop.pipeline.config.PipelineRunConfiguration;
import org.apache.hop.pipeline.engines.EmptyPipelineRunConfiguration;
@GuiPlugin(description = "Local pipeline run configuration widgets")
+@Getter
+@Setter
public class LocalPipelineRunConfiguration extends
EmptyPipelineRunConfiguration
- implements IPipelineEngineRunConfiguration {
+ implements IPipelineEngineRunConfiguration,
+
org.apache.hop.pipeline.engines.IMeasuringLocalPipelineRunConfiguration {
@GuiWidgetElement(
id = "rowSetSize",
@@ -188,164 +193,4 @@ public class LocalPipelineRunConfiguration extends
EmptyPipelineRunConfiguration
}
return list;
}
-
- /**
- * Gets rowSetSize
- *
- * @return value of rowSetSize
- */
- public String getRowSetSize() {
- return rowSetSize;
- }
-
- /**
- * @param rowSetSize The rowSetSize to set
- */
- public void setRowSetSize(String rowSetSize) {
- this.rowSetSize = rowSetSize;
- }
-
- /**
- * Gets safeModeEnabled
- *
- * @return value of safeModeEnabled
- */
- public boolean isSafeModeEnabled() {
- return safeModeEnabled;
- }
-
- /**
- * @param safeModeEnabled The safeModeEnabled to set
- */
- public void setSafeModeEnabled(boolean safeModeEnabled) {
- this.safeModeEnabled = safeModeEnabled;
- }
-
- /**
- * Gets gatheringMetrics
- *
- * @return value of gatheringMetrics
- */
- public boolean isGatheringMetrics() {
- return gatheringMetrics;
- }
-
- /**
- * @param gatheringMetrics The gatheringMetrics to set
- */
- public void setGatheringMetrics(boolean gatheringMetrics) {
- this.gatheringMetrics = gatheringMetrics;
- }
-
- /**
- * Gets sortingTransformsTopologically
- *
- * @return value of sortingTransformsTopologically
- */
- public boolean isSortingTransformsTopologically() {
- return sortingTransformsTopologically;
- }
-
- /**
- * @param sortingTransformsTopologically The sortingTransformsTopologically
to set
- */
- public void setSortingTransformsTopologically(boolean
sortingTransformsTopologically) {
- this.sortingTransformsTopologically = sortingTransformsTopologically;
- }
-
- /**
- * Gets feedbackShown
- *
- * @return value of feedbackShown
- */
- public boolean isFeedbackShown() {
- return feedbackShown;
- }
-
- /**
- * @param feedbackShown The feedbackShown to set
- */
- public void setFeedbackShown(boolean feedbackShown) {
- this.feedbackShown = feedbackShown;
- }
-
- /**
- * Gets feedbackSize
- *
- * @return value of feedbackSize
- */
- public String getFeedbackSize() {
- return feedbackSize;
- }
-
- /**
- * @param feedbackSize The feedbackSize to set
- */
- public void setFeedbackSize(String feedbackSize) {
- this.feedbackSize = feedbackSize;
- }
-
- /**
- * Gets waitTime
- *
- * @return value of the waitTime
- */
- public String getWaitTime() {
- return waitTime;
- }
-
- /**
- * @param waitTime The waitTime for buffer check
- */
- public void setWaitTime(String waitTime) {
- this.waitTime = waitTime;
- }
-
- /**
- * Gets sampleTypeInGui
- *
- * @return value of sampleTypeInGui
- */
- public String getSampleTypeInGui() {
- return sampleTypeInGui;
- }
-
- /**
- * @param sampleTypeInGui The sampleTypeInGui to set
- */
- public void setSampleTypeInGui(String sampleTypeInGui) {
- this.sampleTypeInGui = sampleTypeInGui;
- }
-
- /**
- * Gets sampleSize
- *
- * @return value of sampleSize
- */
- public String getSampleSize() {
- return sampleSize;
- }
-
- /**
- * @param sampleSize The sampleSize to set
- */
- public void setSampleSize(String sampleSize) {
- this.sampleSize = sampleSize;
- }
-
- /**
- * Gets transactional
- *
- * @return value of transactional
- */
- public boolean isTransactional() {
- return transactional;
- }
-
- /**
- * @param transactional The transactional to set
- */
- public void setTransactional(boolean transactional) {
- this.transactional = transactional;
- }
}
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engines/localsingle/LocalSinglePipelineRunConfiguration.java
b/engine/src/main/java/org/apache/hop/pipeline/engines/localsingle/LocalSinglePipelineRunConfiguration.java
deleted file mode 100644
index a507242295..0000000000
---
a/engine/src/main/java/org/apache/hop/pipeline/engines/localsingle/LocalSinglePipelineRunConfiguration.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.hop.pipeline.engines.localsingle;
-
-import org.apache.hop.core.gui.plugin.GuiPlugin;
-import org.apache.hop.pipeline.config.IPipelineEngineRunConfiguration;
-import org.apache.hop.pipeline.engines.EmptyPipelineRunConfiguration;
-
-@GuiPlugin
-public class LocalSinglePipelineRunConfiguration extends
EmptyPipelineRunConfiguration
- implements IPipelineEngineRunConfiguration {
-
- public LocalSinglePipelineRunConfiguration() {
- super();
- }
-
- public LocalSinglePipelineRunConfiguration(
- String pluginId, String pluginName, String rowSetSize) {
- super(pluginId, pluginName);
- }
-
- public
LocalSinglePipelineRunConfiguration(LocalSinglePipelineRunConfiguration config)
{
- super(config);
- }
-
- @Override
- public LocalSinglePipelineRunConfiguration clone() {
- return new LocalSinglePipelineRunConfiguration(this);
- }
-}
diff --git
a/integration-tests/kafka/0002-kafka-consumer-called-subpipeline-with-mapping.hpl
b/integration-tests/kafka/0002-kafka-consumer-called-subpipeline-with-mapping.hpl
new file mode 100644
index 0000000000..4acf0d0848
--- /dev/null
+++
b/integration-tests/kafka/0002-kafka-consumer-called-subpipeline-with-mapping.hpl
@@ -0,0 +1,154 @@
+<?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>
+ <pipeline_version/>
+ <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>
+ <pipeline_type>Normal</pipeline_type>
+ <pipeline_status>0</pipeline_status>
+ <parameters/>
+ <name>0002-kafka-consumer-called-subpipeline-with-mapping</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/21 09:43:31.432</created_date>
+ <modified_date>2021/12/21 09:43:31.432</modified_date>
+ </info>
+ <transform>
+ <type>Dummy</type>
+ <name>Out record</name>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>544</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>Injector</type>
+ <name>Injector</name>
+ <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>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>240</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>SimpleMapping</type>
+ <name>Mapping</name>
+ <runConfiguration>local</runConfiguration>
+ <mappings>
+ <input>
+ <mapping>
+ <connector>
+ <parent>message</parent>
+ <child>in_msg</child>
+ </connector>
+ <main_path>Y</main_path>
+ <rename_on_output>N</rename_on_output>
+ </mapping>
+ </input>
+ <output>
+ <mapping>
+ <connector>
+ <parent>out_msg</parent>
+ <child>Message</child>
+ </connector>
+ <main_path>Y</main_path>
+ <rename_on_output>N</rename_on_output>
+ </mapping>
+ </output>
+ <parameters>
+ <inherit_all_vars>Y</inherit_all_vars>
+ </parameters>
+ </mappings>
+ <filename>${PROJECT_HOME}/0002-mapping-subpipeline.hpl</filename>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>384</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <order>
+ <hop>
+ <from>Injector</from>
+ <to>Mapping</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Mapping</from>
+ <to>Out record</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <notepads/>
+ <attributes/>
+ <transform_error_handling/>
+</pipeline>
diff --git
a/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
b/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
new file mode 100644
index 0000000000..93d136cc7d
--- /dev/null
+++ b/integration-tests/kafka/0002-kafka-consumer-read-record-mapping.hpl
@@ -0,0 +1,166 @@
+<?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>
+ <pipeline_version/>
+ <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>
+ <pipeline_type>Normal</pipeline_type>
+ <pipeline_status>0</pipeline_status>
+ <parameters/>
+ <name>0002-kafka-consumer-read-record-mapping</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/21 09:42:58.199</created_date>
+ <modified_date>2021/12/21 09:42:58.199</modified_date>
+ </info>
+ <transform>
+ <type>KafkaConsumer</type>
+ <name>Kafka Consumer</name>
+ <keyField>
+ <outputName>Key</outputName>
+ <type>String</type>
+ </keyField>
+ <messageField>
+ <outputName>Message</outputName>
+ <type>String</type>
+ </messageField>
+ <topicField>
+ <outputName>Topic</outputName>
+ </topicField>
+ <offsetField>
+ <outputName>Offset</outputName>
+ </offsetField>
+ <partitionField>
+ <outputName>Partition</outputName>
+ </partitionField>
+ <timestampField>
+ <outputName>Timestamp</outputName>
+ </timestampField>
+
<pipelinePath>${PROJECT_HOME}/0002-kafka-consumer-called-subpipeline-with-mapping.hpl</pipelinePath>
+ <executionInformationLocation/>
+ <executionDataProfile/>
+ <batchSize>1</batchSize>
+ <batchDuration>1000</batchDuration>
+ <subTransform>Out record</subTransform>
+ <directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
+ <topic>hop-test-0002</topic>
+ <consumerGroup/>
+ <AUTO_COMMIT>N</AUTO_COMMIT>
+ <options>
+ <option>
+ <property>auto.offset.reset</property>
+ <value>earliest</value>
+ </option>
+ <option>
+ <property>ssl.key.password</property>
+ </option>
+ <option>
+ <property>ssl.keystore.location</property>
+ </option>
+ <option>
+ <property>ssl.keystore.password</property>
+ </option>
+ <option>
+ <property>ssl.truststore.location</property>
+ </option>
+ <option>
+ <property>ssl.truststore.password</property>
+ </option>
+ </options>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>448</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>WriteToLog</type>
+ <name>Log Output</name>
+ <displayHeader>Y</displayHeader>
+ <limitRows>N</limitRows>
+ <limitRowsNumber>0</limitRowsNumber>
+ <logmessage>Kafka Output</logmessage>
+ <loglevel>Basic</loglevel>
+ <fields>
+ <field>
+ <name>message</name>
+ </field>
+ </fields>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>560</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ <description/>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </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/>
+ <transform_error_handling/>
+</pipeline>
diff --git a/integration-tests/kafka/0002-mapping-subpipeline.hpl
b/integration-tests/kafka/0002-mapping-subpipeline.hpl
new file mode 100644
index 0000000000..eaec9bc83c
--- /dev/null
+++ b/integration-tests/kafka/0002-mapping-subpipeline.hpl
@@ -0,0 +1,134 @@
+<?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>0002-mapping-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>2022/11/08 21:16:00.171</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2022/11/08 21:16:00.171</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Input</from>
+ <to>Map Msg</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Map Msg</from>
+ <to>Output</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Input</name>
+ <type>MappingInput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <name>in_msg</name>
+ <type>String</type>
+ <length>-1</length>
+ <precision>-1</precision>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Output</name>
+ <type>MappingOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>480</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Map Msg</name>
+ <type>ScriptValueMod</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <optimizationLevel>9</optimizationLevel>
+ <jsScripts>
+ <jsScript>
+ <jsScript_type>0</jsScript_type>
+ <jsScript_name>Script 1</jsScript_name>
+ <jsScript_script>
+var out_msg = "Mapped: " + in_msg;
+</jsScript_script>
+ </jsScript>
+ </jsScripts>
+ <fields>
+ <field>
+ <name>out_msg</name>
+ <rename>out_msg</rename>
+ <type>String</type>
+ <length>-1</length>
+ <precision>-1</precision>
+ <replace>N</replace>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>288</xloc>
+ <yloc>80</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/kafka/disabled.txt
b/integration-tests/kafka/disabled.txt
deleted file mode 100644
index 7560e5b3bc..0000000000
--- a/integration-tests/kafka/disabled.txt
+++ /dev/null
@@ -1,17 +0,0 @@
-#
-# 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.
-#
-#
\ No newline at end of file
diff --git a/integration-tests/kafka/main-0001-kafka-test-basic.hwf
b/integration-tests/kafka/main-0001-kafka-test-basic.hwf
index 6889b3b75d..f85098a0e2 100644
--- a/integration-tests/kafka/main-0001-kafka-test-basic.hwf
+++ b/integration-tests/kafka/main-0001-kafka-test-basic.hwf
@@ -22,71 +22,67 @@ limitations under the License.
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
- <workflow_version/>
- <workflow_status>0</workflow_status>
<created_user>-</created_user>
- <created_date>2021/12/30 10:42:13.629</created_date>
<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>
- <default_value/>
<description/>
+ <default_value/>
</parameter>
</parameters>
<actions>
<action>
- <name>Start</name>
- <description/>
- <type>SPECIAL</type>
- <attributes/>
<repeat>N</repeat>
<schedulerType>0</schedulerType>
<intervalSeconds>0</intervalSeconds>
<intervalMinutes>60</intervalMinutes>
- <hour>12</hour>
- <minutes>0</minutes>
- <weekDay>1</weekDay>
<DayOfMonth>1</DayOfMonth>
- <parallel>N</parallel>
+ <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>
- <name>prepare-kafka-test-basic.hpl</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
<filename>${PROJECT_HOME}/prepare-kafka-test-basic.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>
- <set_append_logfile>N</set_append_logfile>
<wait_until_finished>Y</wait_until_finished>
- <create_parent_folder>N</create_parent_folder>
- <run_configuration>local</run_configuration>
<parameters>
<pass_all_parameters>Y</pass_all_parameters>
</parameters>
- <parallel>N</parallel>
- <xloc>256</xloc>
+ <run_configuration>local</run_configuration>
+ <name>prepare-kafka-test-basic.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>336</xloc>
<yloc>144</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>Check log</name>
- <description/>
- <type>EVAL</type>
- <attributes/>
<script>
var txt = previous_result.getLogText();
@@ -109,37 +105,41 @@ if (!ok) {
}
ok;</script>
- <parallel>N</parallel>
- <xloc>576</xloc>
+ <name>Check log</name>
+ <description/>
+ <type>EVAL</type>
+ <attributes/>
+ <xloc>864</xloc>
<yloc>144</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>0001-kafka-consumer-read-record-basic.hpl</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
<filename>${PROJECT_HOME}/0001-kafka-consumer-read-record-basic.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>
- <set_append_logfile>N</set_append_logfile>
<wait_until_finished>Y</wait_until_finished>
- <create_parent_folder>N</create_parent_folder>
- <run_configuration>local</run_configuration>
<parameters>
<pass_all_parameters>Y</pass_all_parameters>
</parameters>
- <parallel>N</parallel>
- <xloc>416</xloc>
+ <run_configuration>local</run_configuration>
+ <name>0001-kafka-consumer-read-record-basic.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>608</xloc>
<yloc>144</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
</actions>
@@ -147,26 +147,25 @@ ok;</script>
<hop>
<from>Start</from>
<to>prepare-kafka-test-basic.hpl</to>
- <enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
<hop>
<from>0001-kafka-consumer-read-record-basic.hpl</from>
<to>Check log</to>
- <enabled>Y</enabled>
<evaluation>N</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
<hop>
<from>prepare-kafka-test-basic.hpl</from>
<to>0001-kafka-consumer-read-record-basic.hpl</to>
- <enabled>Y</enabled>
<evaluation>N</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
</hops>
- <notepads>
- </notepads>
+ <notepads/>
<attributes/>
</workflow>
diff --git a/integration-tests/kafka/main-0001-kafka-test-basic.hwf
b/integration-tests/kafka/main-0002-kafka-test-mapping.hwf
similarity index 86%
copy from integration-tests/kafka/main-0001-kafka-test-basic.hwf
copy to integration-tests/kafka/main-0002-kafka-test-mapping.hwf
index 6889b3b75d..e9ba2175a6 100644
--- a/integration-tests/kafka/main-0001-kafka-test-basic.hwf
+++ b/integration-tests/kafka/main-0002-kafka-test-mapping.hwf
@@ -18,82 +18,78 @@ limitations under the License.
-->
<workflow>
- <name>main-0001-kafka-test-basic</name>
+ <name>main-0002-kafka-test-mapping</name>
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
- <workflow_version/>
- <workflow_status>0</workflow_status>
<created_user>-</created_user>
- <created_date>2021/12/30 10:42:13.629</created_date>
<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>
- <default_value/>
<description/>
+ <default_value/>
</parameter>
</parameters>
<actions>
<action>
- <name>Start</name>
- <description/>
- <type>SPECIAL</type>
- <attributes/>
<repeat>N</repeat>
<schedulerType>0</schedulerType>
<intervalSeconds>0</intervalSeconds>
<intervalMinutes>60</intervalMinutes>
- <hour>12</hour>
- <minutes>0</minutes>
- <weekDay>1</weekDay>
<DayOfMonth>1</DayOfMonth>
- <parallel>N</parallel>
+ <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>
- <name>prepare-kafka-test-basic.hpl</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
- <filename>${PROJECT_HOME}/prepare-kafka-test-basic.hpl</filename>
+ <filename>${PROJECT_HOME}/prepare-kafka-test-0002.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>
- <set_append_logfile>N</set_append_logfile>
<wait_until_finished>Y</wait_until_finished>
- <create_parent_folder>N</create_parent_folder>
- <run_configuration>local</run_configuration>
<parameters>
<pass_all_parameters>Y</pass_all_parameters>
</parameters>
- <parallel>N</parallel>
- <xloc>256</xloc>
+ <run_configuration>local</run_configuration>
+ <name>prepare-kafka-test-0002.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>336</xloc>
<yloc>144</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>Check log</name>
- <description/>
- <type>EVAL</type>
- <attributes/>
<script>
var txt = previous_result.getLogText();
var ok = false;
var expectedValues = [
- "Message = Hello Hop!",
+ "Message = Mapped: Hello Hop Mapping!",
];
for (var i = 0 ; i<expectedValues.length ; i++) {
@@ -109,64 +105,67 @@ if (!ok) {
}
ok;</script>
- <parallel>N</parallel>
- <xloc>576</xloc>
+ <name>Check log</name>
+ <description/>
+ <type>EVAL</type>
+ <attributes/>
+ <xloc>864</xloc>
<yloc>144</yloc>
+ <parallel>N</parallel>
<attributes_hac/>
</action>
<action>
- <name>0001-kafka-consumer-read-record-basic.hpl</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
-
<filename>${PROJECT_HOME}/0001-kafka-consumer-read-record-basic.hpl</filename>
+
<filename>${PROJECT_HOME}/0002-kafka-consumer-read-record-mapping.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>
- <set_append_logfile>N</set_append_logfile>
<wait_until_finished>Y</wait_until_finished>
- <create_parent_folder>N</create_parent_folder>
- <run_configuration>local</run_configuration>
<parameters>
<pass_all_parameters>Y</pass_all_parameters>
</parameters>
- <parallel>N</parallel>
- <xloc>416</xloc>
+ <run_configuration>local</run_configuration>
+ <name>0002-kafka-consumer-read-record-mapping.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-basic.hpl</to>
- <enabled>Y</enabled>
+ <to>prepare-kafka-test-0002.hpl</to>
<evaluation>Y</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
<hop>
- <from>0001-kafka-consumer-read-record-basic.hpl</from>
+ <from>0002-kafka-consumer-read-record-mapping.hpl</from>
<to>Check log</to>
- <enabled>Y</enabled>
<evaluation>N</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
<hop>
- <from>prepare-kafka-test-basic.hpl</from>
- <to>0001-kafka-consumer-read-record-basic.hpl</to>
- <enabled>Y</enabled>
+ <from>prepare-kafka-test-0002.hpl</from>
+ <to>0002-kafka-consumer-read-record-mapping.hpl</to>
<evaluation>N</evaluation>
<unconditional>Y</unconditional>
+ <enabled>Y</enabled>
</hop>
</hops>
- <notepads>
- </notepads>
+ <notepads/>
<attributes/>
</workflow>
diff --git a/integration-tests/kafka/prepare-kafka-test-0002.hpl
b/integration-tests/kafka/prepare-kafka-test-0002.hpl
new file mode 100644
index 0000000000..2c269fdc78
--- /dev/null
+++ b/integration-tests/kafka/prepare-kafka-test-0002.hpl
@@ -0,0 +1,113 @@
+<?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-0002</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:37:38.673</created_date>
+ <modified_user>-</modified_user>
+ <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>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Kafka Producer</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Generate rows</name>
+ <type>RowGenerator</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <length>-1</length>
+ <name>field</name>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <type>String</type>
+ <nullif>Hello Hop Mapping!</nullif>
+ </field>
+ </fields>
+ <interval_in_ms>5000</interval_in_ms>
+ <never_ending>N</never_ending>
+ <limit>1</limit>
+ <row_time_field>now</row_time_field>
+ <attributes/>
+ <GUI>
+ <xloc>288</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Kafka Producer</name>
+ <type>KafkaProducerOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <directBootstrapServers>${BOOTSTRAP_SERVERS}</directBootstrapServers>
+ <topic>hop-test-0002</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>464</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/plugins/engines/pom.xml b/plugins/engines/pom.xml
index e9a61a70f3..e03f28ca50 100644
--- a/plugins/engines/pom.xml
+++ b/plugins/engines/pom.xml
@@ -31,6 +31,7 @@
<modules>
<module>beam</module>
+ <module>single-threaded</module>
</modules>
</project>
diff --git a/plugins/engines/pom.xml b/plugins/engines/single-threaded/pom.xml
similarity index 64%
copy from plugins/engines/pom.xml
copy to plugins/engines/single-threaded/pom.xml
index e9a61a70f3..8933ebf359 100644
--- a/plugins/engines/pom.xml
+++ b/plugins/engines/single-threaded/pom.xml
@@ -21,16 +21,26 @@
<parent>
<groupId>org.apache.hop</groupId>
- <artifactId>hop-plugins</artifactId>
+ <artifactId>hop-plugins-engines</artifactId>
<version>2.19.0-SNAPSHOT</version>
</parent>
- <artifactId>hop-plugins-engines</artifactId>
- <packaging>pom</packaging>
- <name>Hop Plugins Engines</name>
-
- <modules>
- <module>beam</module>
- </modules>
+ <artifactId>hop-engines-single-threaded</artifactId>
+ <packaging>jar</packaging>
+ <name>Hop Plugins Engines Single Threaded</name>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-engine</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
</project>
diff --git a/plugins/engines/pom.xml
b/plugins/engines/single-threaded/src/assembly/assembly.xml
similarity index 54%
copy from plugins/engines/pom.xml
copy to plugins/engines/single-threaded/src/assembly/assembly.xml
index e9a61a70f3..74b5ab580f 100644
--- a/plugins/engines/pom.xml
+++ b/plugins/engines/single-threaded/src/assembly/assembly.xml
@@ -1,4 +1,3 @@
-<?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
@@ -14,23 +13,22 @@
~ 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.
- ~
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.hop</groupId>
- <artifactId>hop-plugins</artifactId>
- <version>2.19.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>hop-plugins-engines</artifactId>
- <packaging>pom</packaging>
- <name>Hop Plugins Engines</name>
-
- <modules>
- <module>beam</module>
- </modules>
-</project>
+<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.2.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.2.0
http://maven.apache.org/xsd/assembly-2.2.0.xsd";>
+ <id>hop-engines-single-threaded</id>
+ <formats>
+ <format>zip</format>
+ </formats>
+ <baseDirectory>.</baseDirectory>
+ <dependencySets>
+ <dependencySet>
+ <includes>
+
<include>org.apache.hop:hop-engines-single-threaded:jar</include>
+ </includes>
+ <outputDirectory>plugins/engines/single-threaded</outputDirectory>
+ </dependencySet>
+ </dependencySets>
+</assembly>
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engines/localsingle/LocalSinglePipelineEngine.java
b/plugins/engines/single-threaded/src/main/java/org/apache/hop/pipeline/engines/singlethreaded/SingleThreadedPipelineEngine.java
similarity index 74%
rename from
engine/src/main/java/org/apache/hop/pipeline/engines/localsingle/LocalSinglePipelineEngine.java
rename to
plugins/engines/single-threaded/src/main/java/org/apache/hop/pipeline/engines/singlethreaded/SingleThreadedPipelineEngine.java
index 323b85a733..0deb7de0b2 100644
---
a/engine/src/main/java/org/apache/hop/pipeline/engines/localsingle/LocalSinglePipelineEngine.java
+++
b/plugins/engines/single-threaded/src/main/java/org/apache/hop/pipeline/engines/singlethreaded/SingleThreadedPipelineEngine.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.hop.pipeline.engines.localsingle;
+package org.apache.hop.pipeline.engines.singlethreaded;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.ILoggingObject;
@@ -28,28 +28,29 @@ import
org.apache.hop.pipeline.SingleThreadedPipelineExecutor;
import org.apache.hop.pipeline.config.IPipelineEngineRunConfiguration;
import org.apache.hop.pipeline.engine.IPipelineEngine;
import org.apache.hop.pipeline.engine.PipelineEngineCapabilities;
+import org.apache.hop.pipeline.engine.PipelineEnginePlugin;
-// @PipelineEnginePlugin(
-// id = "LocalSingle",
-// name = "Hop local single threaded pipeline engine",
-// description = "Executes your pipeline locally in a single-threaded fashion"
-// )
-public class LocalSinglePipelineEngine extends Pipeline implements
IPipelineEngine<PipelineMeta> {
+@PipelineEnginePlugin(
+ id = "LocalSingle",
+ name = "Hop local single threaded pipeline engine",
+ description = "Executes your pipeline locally in a single-threaded
fashion")
+public class SingleThreadedPipelineEngine extends Pipeline
+ implements IPipelineEngine<PipelineMeta> {
- public LocalSinglePipelineEngine() {
+ public SingleThreadedPipelineEngine() {
super();
}
- public LocalSinglePipelineEngine(PipelineMeta pipelineMeta) {
+ public SingleThreadedPipelineEngine(PipelineMeta pipelineMeta) {
super(pipelineMeta);
}
- public LocalSinglePipelineEngine(
+ public SingleThreadedPipelineEngine(
PipelineMeta pipelineMeta, IVariables variables, ILoggingObject parent) {
super(pipelineMeta, variables, parent);
}
- public <Parent extends IVariables & INamedParameters>
LocalSinglePipelineEngine(
+ public <Parent extends IVariables & INamedParameters>
SingleThreadedPipelineEngine(
Parent parent, String name, String filename, IHopMetadataProvider
metadataProvider)
throws HopException {
super(parent, name, filename, metadataProvider);
@@ -57,14 +58,9 @@ public class LocalSinglePipelineEngine extends Pipeline
implements IPipelineEngi
@Override
public IPipelineEngineRunConfiguration
createDefaultPipelineEngineRunConfiguration() {
- return new LocalSinglePipelineRunConfiguration();
+ return new SingleThreadedPipelineRunConfiguration();
}
- /**
- * Should support everything
- *
- * @return
- */
@Override
public PipelineEngineCapabilities getEngineCapabilities() {
return new PipelineEngineCapabilities(true, true, true, true);
@@ -87,19 +83,19 @@ public class LocalSinglePipelineEngine extends Pipeline
implements IPipelineEngi
"Error initializing single threaded pipeline execution. See the log
for more details.");
}
- // Iterate until done.
+ // One iteration will execute each transform once.
//
- while (executor.oneIteration() && !isStopped())
- ;
+ try {
+ executor.oneIteration();
+ } finally {
+ // All data should now be processed.
+ //
+ executor.dispose();
+ }
}
@Override
public String getStatusDescription() {
return super.getStatus();
}
-
- @Override
- public void pipelineCompleted() throws HopException {
- // Do nothing
- }
}
diff --git
a/plugins/engines/single-threaded/src/main/java/org/apache/hop/pipeline/engines/singlethreaded/SingleThreadedPipelineRunConfiguration.java
b/plugins/engines/single-threaded/src/main/java/org/apache/hop/pipeline/engines/singlethreaded/SingleThreadedPipelineRunConfiguration.java
new file mode 100644
index 0000000000..025e8007e1
--- /dev/null
+++
b/plugins/engines/single-threaded/src/main/java/org/apache/hop/pipeline/engines/singlethreaded/SingleThreadedPipelineRunConfiguration.java
@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+
+package org.apache.hop.pipeline.engines.singlethreaded;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.core.gui.plugin.GuiElementType;
+import org.apache.hop.core.gui.plugin.GuiPlugin;
+import org.apache.hop.core.gui.plugin.GuiWidgetElement;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.config.IPipelineEngineRunConfiguration;
+import org.apache.hop.pipeline.config.PipelineRunConfiguration;
+import org.apache.hop.pipeline.engines.EmptyPipelineRunConfiguration;
+import org.apache.hop.pipeline.engines.IMeasuringLocalPipelineRunConfiguration;
+import org.apache.hop.pipeline.engines.local.LocalPipelineRunConfiguration;
+
+@GuiPlugin
+@Getter
+@Setter
+public class SingleThreadedPipelineRunConfiguration extends
EmptyPipelineRunConfiguration
+ implements IPipelineEngineRunConfiguration,
IMeasuringLocalPipelineRunConfiguration {
+ /** The feedback size. */
+ @GuiWidgetElement(
+ id = "sampleTypeInGui",
+ order = "080",
+ parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.COMBO,
+ label =
+
"i18n:org.apache.hop.ui.pipeline.config:PipelineRunConfigurationDialog.SampleTypeInGui.Label",
+ comboValuesMethod = "getSampleTypes")
+ @HopMetadataProperty(key = "sample_type_in_gui")
+ protected String sampleTypeInGui;
+
+ /** The feedback size. */
+ @GuiWidgetElement(
+ id = "sampleSize",
+ order = "090",
+ parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label =
+
"i18n:org.apache.hop.ui.pipeline.config:PipelineRunConfigurationDialog.SampleSize.Label")
+ @HopMetadataProperty(key = "sample_size")
+ protected String sampleSize;
+
+ public SingleThreadedPipelineRunConfiguration() {
+ super();
+ this.sampleTypeInGui =
LocalPipelineRunConfiguration.SampleType.Last.name();
+ this.sampleSize = "100";
+ }
+
+ public SingleThreadedPipelineRunConfiguration(
+ String pluginId, String pluginName, String rowSetSize) {
+ super(pluginId, pluginName);
+ }
+
+ public
SingleThreadedPipelineRunConfiguration(SingleThreadedPipelineRunConfiguration
config) {
+ super(config);
+ this.sampleTypeInGui = config.sampleTypeInGui;
+ this.sampleSize = config.sampleSize;
+ }
+
+ @Override
+ public SingleThreadedPipelineRunConfiguration clone() {
+ return new SingleThreadedPipelineRunConfiguration(this);
+ }
+
+ public List<String> getSampleTypes(ILogChannel log, IHopMetadataProvider
metadataProvider) {
+ List<String> list = new ArrayList<>();
+ for (LocalPipelineRunConfiguration.SampleType type :
+ LocalPipelineRunConfiguration.SampleType.values()) {
+ list.add(type.name());
+ }
+ return list;
+ }
+}
diff --git
a/plugins/misc/testing/src/main/java/org/apache/hop/testing/xp/InjectDataSetIntoTransformExtensionPoint.java
b/plugins/misc/testing/src/main/java/org/apache/hop/testing/xp/InjectDataSetIntoTransformExtensionPoint.java
index 32c4599d18..02845c7f91 100644
---
a/plugins/misc/testing/src/main/java/org/apache/hop/testing/xp/InjectDataSetIntoTransformExtensionPoint.java
+++
b/plugins/misc/testing/src/main/java/org/apache/hop/testing/xp/InjectDataSetIntoTransformExtensionPoint.java
@@ -23,7 +23,6 @@ import java.util.List;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.exception.HopException;
-import org.apache.hop.core.exception.HopPluginException;
import org.apache.hop.core.exception.HopRuntimeException;
import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.extension.ExtensionPoint;
@@ -62,10 +61,6 @@ public class InjectDataSetIntoTransformExtensionPoint
ILogChannel log, IVariables variables, final
IPipelineEngine<PipelineMeta> pipeline)
throws HopException {
- if (!(pipeline instanceof LocalPipelineEngine)) {
- throw new HopPluginException("Unit tests can only run using a local
pipeline engine type");
- }
-
final PipelineMeta pipelineMeta = pipeline.getPipelineMeta();
boolean dataSetEnabled =
"Y".equalsIgnoreCase(pipeline.getVariable(DataSetConst.VAR_RUN_UNIT_TEST));
@@ -76,6 +71,11 @@ public class InjectDataSetIntoTransformExtensionPoint
return;
}
+ if (!(pipeline instanceof LocalPipelineEngine)) {
+ log.logBasic("Unit tests can only run using a local pipeline engine
type");
+ return;
+ }
+
String unitTestName =
pipeline.getVariable(DataSetConst.VAR_UNIT_TEST_NAME);
if (log.isDetailed()) {
log.logDetailed("Unit test name: " + unitTestName);
diff --git
a/plugins/transforms/mapping/src/main/java/org/apache/hop/pipeline/transforms/mapping/SimpleMapping.java
b/plugins/transforms/mapping/src/main/java/org/apache/hop/pipeline/transforms/mapping/SimpleMapping.java
index 06052bdd35..74dea6871d 100644
---
a/plugins/transforms/mapping/src/main/java/org/apache/hop/pipeline/transforms/mapping/SimpleMapping.java
+++
b/plugins/transforms/mapping/src/main/java/org/apache/hop/pipeline/transforms/mapping/SimpleMapping.java
@@ -122,7 +122,13 @@ public class SimpleMapping extends
BaseTransform<SimpleMappingMeta, SimpleMappin
}
public void prepareMappingExecution() throws HopException {
- if (data.isBeamContext()) {
+ boolean singleThreaded =
+ data.isBeamContext()
+ || (getPipeline() != null
+ && getPipeline().getPipelineMeta() != null
+ && getPipeline().getPipelineMeta().getPipelineType()
+ == PipelineMeta.PipelineType.SingleThreaded);
+ if (singleThreaded) {
data.mappingPipelineMeta.setPipelineType(PipelineMeta.PipelineType.SingleThreaded);
}
SimpleMappingData simpleMappingData = getData();
@@ -286,7 +292,13 @@ public class SimpleMapping extends
BaseTransform<SimpleMappingMeta, SimpleMappin
// We don't want to process one-row batches in a parallel engine
where we need to wait for
// the threads to finish.
//
- if (data.isBeamContext()) {
+ boolean singleThreaded =
+ data.isBeamContext()
+ || (getPipeline() != null
+ && getPipeline().getPipelineMeta() != null
+ && getPipeline().getPipelineMeta().getPipelineType()
+ == PipelineMeta.PipelineType.SingleThreaded);
+ if (singleThreaded) {
data.executor = new
SingleThreadedPipelineExecutor(data.mappingPipeline);
}
@@ -306,7 +318,7 @@ public class SimpleMapping extends
BaseTransform<SimpleMappingMeta, SimpleMappin
@Override
public void dispose() {
- if (data.isBeamContext()) {
+ if (data.executor != null) {
try {
data.executor.dispose();
} catch (Exception e) {
@@ -332,7 +344,7 @@ public class SimpleMapping extends
BaseTransform<SimpleMappingMeta, SimpleMappin
@Override
public void batchComplete() throws HopException {
- if (data.isBeamContext()) {
+ if (data.executor != null) {
// Execute all transforms single-threaded, one after the other.
// This way input rows can end up in the mapping output transform.
//
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
index 9a3944e8e4..f8bd920cfa 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
@@ -31,8 +31,8 @@ import org.apache.hop.core.row.RowBuffer;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.engine.IEngineComponent;
import org.apache.hop.pipeline.engine.IPipelineEngine;
+import org.apache.hop.pipeline.engines.IMeasuringLocalPipelineRunConfiguration;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
-import org.apache.hop.pipeline.engines.local.LocalPipelineRunConfiguration;
import
org.apache.hop.pipeline.engines.local.LocalPipelineRunConfiguration.SampleType;
import org.apache.hop.pipeline.transform.RowAdapter;
@@ -55,17 +55,15 @@ public final class PipelineRowSamplerHelper {
*/
public static void addRowSamplersToPipeline(
IPipelineEngine<PipelineMeta> pipeline, Map<String, RowBuffer>
targetMap) {
- if (pipeline == null || targetMap == null || !(pipeline instanceof
LocalPipelineEngine)) {
+ if (pipeline == null || targetMap == null) {
return;
}
+
if (pipeline.getPipelineRunConfiguration() == null
|| !(pipeline.getPipelineRunConfiguration().getEngineRunConfiguration()
- instanceof LocalPipelineRunConfiguration)) {
+ instanceof IMeasuringLocalPipelineRunConfiguration lprConfig)) {
return;
}
- LocalPipelineRunConfiguration lprConfig =
- (LocalPipelineRunConfiguration)
- pipeline.getPipelineRunConfiguration().getEngineRunConfiguration();
if (StringUtils.isEmpty(lprConfig.getSampleTypeInGui())) {
return;
}
