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 bab67a10d0 Issue apache#2340 [Bug]: Beam - WriteToLog never logs when
running via Beam-Direct (#8127)
bab67a10d0 is described below
commit bab67a10d01b76e6f93f30dde735d50fc87c1b04
Author: Sergio Ramazzina <[email protected]>
AuthorDate: Thu Aug 27 16:33:42 2026 +0200
Issue apache#2340 [Bug]: Beam - WriteToLog never logs when running via
Beam-Direct (#8127)
- The converter reloaded the run configuration from the metadata, so the
runtime
log-level variable was missing and the level fell back to MINIMAL,
swallowing
BASIC-level messages: default to BASIC instead.
- The per-transform pipeline's log channel was attached to a detached
logging
object, so its messages never reached the parent pipeline log: attach it
to the
parent Beam pipeline log channel when present, falling back to a
standalone
logging object otherwise.
- Added a Beam Direct unit test and a Beam Direct integration test covering
the
Write to log transform.
Fixes #2340
---
.../beam_directrunner/0013-write-to-log.hpl | 100 +++++++++++++++++
.../beam_directrunner/main-0013-write-to-log.hwf | 124 +++++++++++++++++++++
plugins/engines/beam/pom.xml | 6 +
.../hop/beam/core/transform/TransformFn.java | 15 ++-
.../HopPipelineMetaToBeamPipelineConverter.java | 9 +-
.../hop/beam/transform/WriteToLogPipelineTest.java | 85 ++++++++++++++
.../apache/hop/beam/util/BeamPipelineMetaUtil.java | 44 ++++++++
7 files changed, 379 insertions(+), 4 deletions(-)
diff --git a/integration-tests/beam_directrunner/0013-write-to-log.hpl
b/integration-tests/beam_directrunner/0013-write-to-log.hpl
new file mode 100644
index 0000000000..628ccae7c1
--- /dev/null
+++ b/integration-tests/beam_directrunner/0013-write-to-log.hpl
@@ -0,0 +1,100 @@
+<?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>
+ <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>-1</pipeline_status>
+ <parameters/>
+ <name>New pipeline</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <created_user>-</created_user>
+ <modified_user>-</modified_user>
+ <created_date>2026/08/27 14:54:21.077</created_date>
+ <modified_date>2026/08/27 14:54:21.077</modified_date>
+ </info>
+ <transform>
+ <type>RowGenerator</type>
+ <name>Generate rows</name>
+ <never_ending>N</never_ending>
+ <interval_in_ms>5000</interval_in_ms>
+ <row_time_field>now</row_time_field>
+ <last_time_field>FiveSecondsAgo</last_time_field>
+ <limit>1</limit>
+ <fields>
+ <field>
+ <name>foo</name>
+ <type>String</type>
+ <format/>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif>Hello Apache Hop!</nullif>
+ <set_empty_string>N</set_empty_string>
+ </field>
+ </fields>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>224</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <transform>
+ <type>WriteToLog</type>
+ <name>Write to log</name>
+ <displayHeader>Y</displayHeader>
+ <limitRows>N</limitRows>
+ <limitRowsNumber>0</limitRowsNumber>
+ <logmessage/>
+ <loglevel>Basic</loglevel>
+ <fields/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <GUI>
+ <xloc>480</xloc>
+ <yloc>160</yloc>
+ </GUI>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ </transform>
+ <order>
+ <hop>
+ <from>Generate rows</from>
+ <to>Write to log</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <notepads/>
+ <attributes/>
+ <transform_error_handling/>
+</pipeline>
diff --git a/integration-tests/beam_directrunner/main-0013-write-to-log.hwf
b/integration-tests/beam_directrunner/main-0013-write-to-log.hwf
new file mode 100644
index 0000000000..14f104fa60
--- /dev/null
+++ b/integration-tests/beam_directrunner/main-0013-write-to-log.hwf
@@ -0,0 +1,124 @@
+<?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-0013-write-to-log</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <created_user>-</created_user>
+ <modified_user>-</modified_user>
+ <created_date>2026/08/27 15:06:07.411</created_date>
+ <modified_date>2026/08/27 15:06:07.411</modified_date>
+ <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>96</xloc>
+ <yloc>176</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <filename>${PROJECT_HOME}/0013-write-to-log.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>
+ <wait_timeout/>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <run_configuration>local</run_configuration>
+ <name>0013-write-to-log.hpl</name>
+ <type>PIPELINE</type>
+ <attributes/>
+ <xloc>272</xloc>
+ <yloc>176</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ <action>
+ <script>
+var txt = previous_result.getLogText();
+
+
+var ok = true;
+
+var expectedValues = [
+ "foo = Hello Apache Hop!"
+ ];
+
+for (var i = 0 ; i<expectedValues.length ; i++) {
+ var expectedValue = expectedValues[i];
+ if ( !txt.contains(expectedValue)) {
+ ok = false;
+ log.logError("Value logged as '" + expectedValue + "' was not logged at
least once");
+ }
+}
+
+ok;</script>
+ <name>Check log</name>
+ <description/>
+ <type>EVAL</type>
+ <attributes/>
+ <xloc>464</xloc>
+ <yloc>176</yloc>
+ <parallel>N</parallel>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>0013-write-to-log.hpl</to>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>0013-write-to-log.hpl</from>
+ <to>Check log</to>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ <enabled>Y</enabled>
+ </hop>
+ </hops>
+ <notepads/>
+ <attributes/>
+</workflow>
diff --git a/plugins/engines/beam/pom.xml b/plugins/engines/beam/pom.xml
index fc6f029a09..215a417533 100644
--- a/plugins/engines/beam/pom.xml
+++ b/plugins/engines/beam/pom.xml
@@ -1643,6 +1643,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-transform-writetolog</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
diff --git
a/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/TransformFn.java
b/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/TransformFn.java
index 4347ca35eb..3500aa7895 100644
---
a/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/TransformFn.java
+++
b/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/TransformFn.java
@@ -38,7 +38,9 @@ import
org.apache.hop.beam.engines.HopPipelineExecutionOptions;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.exception.HopRuntimeException;
import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.logging.LoggingObject;
+import org.apache.hop.core.logging.LoggingRegistry;
import org.apache.hop.core.metadata.SerializableMetadataProvider;
import org.apache.hop.core.plugins.PluginRegistry;
import org.apache.hop.core.plugins.TransformPluginType;
@@ -334,9 +336,16 @@ public class TransformFn extends TransformBaseFn {
// Create the transformation...
//
- pipeline =
- new LocalPipelineEngine(
- pipelineMeta, variables, new
LoggingObject("apache-beam-transform"));
+ // Attach the per-transform pipeline's log channel to the parent Beam
pipeline channel so that
+ // messages logged by the wrapped transform (e.g. the "Write to log"
transform) show up in the
+ // pipeline log instead of a detached channel. Fall back to a standalone
logging object when the
+ // parent isn't in the local registry (e.g. on a distributed runner
running in another JVM).
+ ILoggingObject parentLoggingObject =
+ LoggingRegistry.getInstance().getLoggingObject(parentLogChannelId);
+ if (parentLoggingObject == null) {
+ parentLoggingObject = new LoggingObject("apache-beam-transform");
+ }
+ pipeline = new LocalPipelineEngine(pipelineMeta, variables,
parentLoggingObject);
pipeline.setLogLevel(
context.getPipelineOptions().as(HopPipelineExecutionOptions.class).getLogLevel());
pipeline.setMetadataProvider(pipelineMeta.getMetadataProvider());
diff --git
a/plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/HopPipelineMetaToBeamPipelineConverter.java
b/plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/HopPipelineMetaToBeamPipelineConverter.java
index d1048c7fda..edf25b0bac 100644
---
a/plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/HopPipelineMetaToBeamPipelineConverter.java
+++
b/plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/HopPipelineMetaToBeamPipelineConverter.java
@@ -292,6 +292,13 @@ public class HopPipelineMetaToBeamPipelineConverter {
pipelineOptions.setJobName(sanitizeJobName(pipelineMeta.getName()));
+ // The log level for the per-transform pipelines executed inside the Beam
workers.
+ // The engine sets this as a runtime variable on its live run
configuration, but this converter
+ // reloads a fresh copy of the run configuration from the metadata, so
that runtime variable is
+ // not present here. When it is absent we default to BASIC (the documented
default of
+ // HopPipelineExecutionOptions and the behaviour of the local pipeline
engine); defaulting to a
+ // lower level such as MINIMAL would silently swallow BASIC-level
transform logging (e.g. the
+ // "Write to log" transform) when running on Beam.
pipelineOptions
.as(HopPipelineExecutionOptions.class)
.setLogLevel(
@@ -299,7 +306,7 @@ public class HopPipelineMetaToBeamPipelineConverter {
Const.NVL(
pipelineRunConfiguration.getVariable(
BeamConst.STRING_LOCAL_PIPELINE_FLAG_LOG_LEVEL),
- "MINIMAL")));
+ LogLevel.BASIC.getCode())));
pipelineOptions.setRunner(runnerClass);
}
diff --git
a/plugins/engines/beam/src/test/java/org/apache/hop/beam/transform/WriteToLogPipelineTest.java
b/plugins/engines/beam/src/test/java/org/apache/hop/beam/transform/WriteToLogPipelineTest.java
new file mode 100644
index 0000000000..690d520990
--- /dev/null
+++
b/plugins/engines/beam/src/test/java/org/apache/hop/beam/transform/WriteToLogPipelineTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.beam.transform;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.hop.beam.util.BeamPipelineMetaUtil;
+import org.apache.hop.core.logging.HopLogStore;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engine.EngineMetrics;
+import org.apache.hop.pipeline.engine.IEngineComponent;
+import org.apache.hop.pipeline.engine.IPipelineEngine;
+import org.apache.hop.pipeline.engine.PipelineEngineFactory;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Reproduction test for issue #2340: the "Write to log" transform produces no
visible output in the
+ * pipeline log when the pipeline is executed on the Beam Direct runner (it
works fine on the local
+ * runner).
+ */
+class WriteToLogPipelineTest extends PipelineTestBase {
+
+ private static final String MARKER = "HOP2340WRITETOLOGMARKER";
+
+ @Test
+ void testWriteToLogAppearsInPipelineLogOnBeamDirect() throws Exception {
+
+ PipelineMeta pipelineMeta =
+ BeamPipelineMetaUtil.generateWriteToLogPipelineMeta(
+ "beam-write-to-log", "INPUT", "WriteToLog", MARKER,
metadataProvider);
+ pipelineMeta.lookupReferencesAfterLoading();
+
+ IPipelineEngine<PipelineMeta> pipeline =
+ PipelineEngineFactory.createPipelineEngine(
+ variables, NAME_RUN_CONFIG, metadataProvider, pipelineMeta);
+ pipeline.execute();
+ pipeline.waitUntilFinished();
+
+ // Guard: make sure rows actually flowed through the WriteToLog transform,
so an empty-input
+ // regression can never masquerade as the #2340 bug.
+ //
+ EngineMetrics metrics = pipeline.getEngineMetrics();
+ Long writeToLogRead = null;
+ for (IEngineComponent component : metrics.getComponents()) {
+ if ("WriteToLog".equals(component.getName())) {
+ writeToLogRead = metrics.getComponentMetric(component,
Pipeline.METRIC_READ);
+ }
+ }
+ if (writeToLogRead != null) {
+ assertTrue(
+ writeToLogRead > 0,
+ "WriteToLog should have read rows on the Beam Direct runner (read="
+ + writeToLogRead
+ + ")");
+ }
+
+ // What the user actually sees for this run: the log scoped to the
pipeline's own log channel.
+ //
+ String pipelineLog =
+ HopLogStore.getAppender().getBuffer(pipeline.getLogChannelId(),
true).toString();
+
+ assertTrue(
+ pipelineLog.contains(MARKER),
+ "The WriteToLog message '"
+ + MARKER
+ + "' should appear in the pipeline log when running on the Beam
Direct runner, "
+ + "but it was absent (issue #2340).");
+ }
+}
diff --git
a/plugins/engines/beam/src/test/java/org/apache/hop/beam/util/BeamPipelineMetaUtil.java
b/plugins/engines/beam/src/test/java/org/apache/hop/beam/util/BeamPipelineMetaUtil.java
index 576ee52cba..be48485839 100644
---
a/plugins/engines/beam/src/test/java/org/apache/hop/beam/util/BeamPipelineMetaUtil.java
+++
b/plugins/engines/beam/src/test/java/org/apache/hop/beam/util/BeamPipelineMetaUtil.java
@@ -24,6 +24,7 @@ import org.apache.hop.beam.transform.PipelineTestBase;
import org.apache.hop.beam.transforms.io.BeamInputMeta;
import org.apache.hop.beam.transforms.io.BeamOutputMeta;
import org.apache.hop.core.Condition;
+import org.apache.hop.core.logging.LogLevel;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.ValueMetaAndData;
import org.apache.hop.metadata.api.IHopMetadataProvider;
@@ -42,6 +43,7 @@ import
org.apache.hop.pipeline.transforms.mergejoin.MergeJoinMeta;
import org.apache.hop.pipeline.transforms.streamlookup.StreamLookupMeta;
import org.apache.hop.pipeline.transforms.switchcase.SwitchCaseMeta;
import org.apache.hop.pipeline.transforms.switchcase.SwitchCaseTarget;
+import org.apache.hop.pipeline.transforms.writetolog.WriteToLogMeta;
public class BeamPipelineMetaUtil {
@@ -93,6 +95,48 @@ public class BeamPipelineMetaUtil {
return pipelineMeta;
}
+ public static PipelineMeta generateWriteToLogPipelineMeta(
+ String pipelineName,
+ String inputTransformName,
+ String writeToLogTransformName,
+ String logMessage,
+ IHopMetadataProvider metadataProvider)
+ throws Exception {
+
+ IHopMetadataSerializer<FileDefinition> serializer =
+ metadataProvider.getSerializer(FileDefinition.class);
+ FileDefinition customerFileDefinition =
createCustomersInputFileDefinition();
+ serializer.save(customerFileDefinition);
+
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName(pipelineName);
+ pipelineMeta.setMetadataProvider(metadataProvider);
+
+ // Input transform reading the customers file
+ //
+ BeamInputMeta beamInputMeta = new BeamInputMeta();
+ beamInputMeta.setInputLocation(PipelineTestBase.INPUT_CUSTOMERS_FILE);
+ beamInputMeta.setFileDefinitionName(customerFileDefinition.getName());
+ TransformMeta beamInputTransformMeta = new
TransformMeta(inputTransformName, beamInputMeta);
+ beamInputTransformMeta.setTransformPluginId("BeamInput");
+ pipelineMeta.addTransform(beamInputTransformMeta);
+
+ // Write to log transform, logging a distinctive message at BASIC level
+ //
+ WriteToLogMeta writeToLogMeta = new WriteToLogMeta();
+ writeToLogMeta.setLogLevel(LogLevel.BASIC);
+ writeToLogMeta.setDisplayHeader(true);
+ writeToLogMeta.setLogMessage(logMessage);
+ TransformMeta writeToLogTransformMeta =
+ new TransformMeta(writeToLogTransformName, writeToLogMeta);
+ writeToLogTransformMeta.setTransformPluginId("WriteToLog");
+ pipelineMeta.addTransform(writeToLogTransformMeta);
+ pipelineMeta.addPipelineHop(
+ new PipelineHopMeta(beamInputTransformMeta, writeToLogTransformMeta));
+
+ return pipelineMeta;
+ }
+
public static PipelineMeta generateBeamGroupByPipelineMeta(
String transname,
String inputTransformName,
