mattcasters commented on issue #8708:
URL: https://github.com/apache/hop/issues/8708#issuecomment-5917554159
# Plan — Apache Hop issue #8708 (umbrella: Beam engine, runners, Beam IO)
## Goal
Take the 18 open sub-issues of umbrella #8708 and, for each one, either land
a working,
tested change on branch `issue-8708` or post a concrete, evidence-backed
decision to close /
decline it — leaving the umbrella issue with every checkbox resolved.
## Current context / assumptions
- **Repo**: `/home/matt/tmp/hop`, branch `issue-8708` (currently identical
to `main`; no commits yet).
- **Umbrella #8708** lists 18 sub-issues: 2040, 2275, 2326, 2337, 2355,
2357, 2358, 2375, 2377,
2378, 2379, 2409, 2416, 2613, 2675, 3131, 3478, 5064. Milestone **2.21**.
- **Beam module** lives at `plugins/engines/beam` (artifactId
`hop-engines-beam`, parent
`hop-plugins-engines`). All 19 sub-issues touch either this module, its
docs, or a
small number of sibling transform plugins.
- **Beam version is pinned to 2.74.0** (`pom.xml:100`,
`<apache-beam.version>`).
- **Verified facts about the pinned Beam version** (checked against the
local `~/.m2` jars —
the implementer does not need to re-derive these):
- `TextIO.Write.withCompression(org.apache.beam.sdk.io.Compression)`
**exists**; `Compression`
has `AUTO, UNCOMPRESSED, GZIP, BZIP2, ZIP, ZSTD, LZO, LZOP, DEFLATE,
SNAPPY`.
- `BigQueryIO.TypedRead` has **`withQueryLocation(String)`** and
**`withoutValidation()`**.
- `org.apache.beam.sdk.extensions.gcp.options.GcpOptions` has
**`getGcpTempLocation()` /
`setGcpTempLocation(String)`** (it lives in
`beam-sdks-java-extensions-google-cloud-platform-core`,
which is already on the module classpath).
- `org.apache.beam.sdk.io.GenerateSequence` **exists** in
`beam-sdks-java-core`.
- `org.apache.beam.sdk.io.parquet.ParquetIO` **exists but is NOT on the
classpath** — it lives
in a separate artifact
`org.apache.beam:beam-sdks-java-io-parquet:2.74.0` (confirmed present
on Maven Central, HTTP 200, contains `ParquetIO$Sink` etc.). Adding it
is a pom change.
- `org.apache.beam.sdk.extensions.avro.io.AvroIO` **is already on the
classpath**
(via `beam-sdks-java-extensions-avro`, transitively present in
`dependency:build-classpath` output).
- `beam-sdks-java-io-mqtt`, `-splunk`, `-snowflake`, `-elasticsearch`,
`-debezium` all exist at
`2.74.0` on Maven Central (HTTP 200 verified) but are **not** on the
classpath.
- **Flink JobID (#3131) is NOT obtainable** from `PipelineResult`. Read the
Beam 2.74 sources:
`FlinkRunnerResult` holds only `{accumulators, runtime}`;
`FlinkPipelineExecutionEnvironment
.executePipeline()` builds it from `JobExecutionResult` and **drops** the
`JobID`; the
detached variant `FlinkDetachedRunnerResult` holds a private `JobClient`
with **no public
getter**. Compare with Dataflow, which *does* expose it
(`DataflowPipelineJob.getJobId()`,
already surfaced in `BeamDataFlowPipelineEngine.DETAIL_DATAFLOW_JOB_ID`).
So #3131 is a
**documentation/answer** issue, not a code issue. Same for #3478, which is
a question about
what partitioning/copies mean on Flink.
- **Kafka Avro + schema registry (#2675)**: the Confluent jars
(`kafka-avro-serializer`,
`kafka-schema-registry-client`, `common-utils`, 7.6.1) *are* on the
compile classpath but are
**not** declared in `plugins/engines/beam/pom.xml` and are **not** in
`plugins/engines/beam/lib`,
so `FatJarBuilder` (which globs `plugins/engines/beam/lib/*.jar` via
`BeamConst.findLibraryFilesToStage`) never stages them → the reported
`ClassNotFoundException:
io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe` on the
Flink task manager. The produce-side NPE
(`AvroCoder$AvroDeterminismChecker.checkIndexedRecord`,
thrown from `BeamProduceMeta.handleTransform:149`) happens **before** any
cluster runs and is a
separate defect in the same area.
- **Build/test commands that work on this checkout** (verified):
- Compile only: `./mvnw -o -q -pl plugins/engines/beam -am -DskipTests
-Dspotless.check.skip=true -Drat.skip=true install` → exit 0.
- Module tests: `./mvnw -o -pl plugins/engines/beam
-Dspotless.check.skip=true -Drat.skip=true -Pskip-uitest test` → `Tests run:
67, Failures: 0, Errors: 0, Skipped: 0` / `BUILD SUCCESS` in ~42 s.
- **House rules that apply to every code task** (from `AGENTS.md` /
`CLAUDE.md`):
- Apache License 2.0 header on every new file (Apache RAT fails otherwise).
- `./mvnw spotless:apply` before commit; CI runs `spotless:check`.
- Lombok `@Getter`/`@Setter` for new classes rather than hand-written
accessors.
- `metadata type key` is kebab-case; `rdbms` is the one grandfathered
exception.
- New `messages_en_US.properties` entries: quote variable examples as
`'${VAR}'`.
- Never read/print `.env` or credentials. Don't commit unless asked — but
this plan's tasks
**do** end in a commit each, per the TDD cadence requested.
- **Test conventions in this module** (mirror them exactly):
- Transform-meta unit tests:
`src/test/java/org/apache/hop/beam/transforms/<pkg>/<X>MetaTest.java`,
use
`TransformSerializationTestUtil.testSerialization("/<x>-transform.xml",
XMeta.class)` with a
fixture in `src/test/resources/`, and `@BeforeAll HopEnvironment.init();
PluginRegistry.init();`.
- End-to-end pipeline tests: extend
`src/test/java/org/apache/hop/beam/transform/PipelineTestBase.java`
(registers plugin classes, builds a `MemoryMetadataProvider`, a `direct`
run config, copies
`src/test/resources/customers/*` into `/tmp/customers/io`).
- **Docs** are Antora asciidoc under `docs/hop-user-manual/modules/ROOT/`,
wired via
`docs/hop-user-manual/modules/ROOT/nav.adoc` (Beam entries at lines 58–66).
## Architecture / proposed approach
Work in **five independently shippable waves**, ordered so that each wave
leaves `main`-green
and every later wave benefits from the shared scaffolding built in wave 1.
Wave 1 builds a
*triage table* (per-issue decision + evidence) and a *pipeline-test
fixture*; wave 2 lands the four
cheap, high-confidence code changes; wave 3 lands the transform-handler
work; wave 4 lands the
five new Beam IO transforms; wave 5 writes the documentation and closes out
the umbrella. Every
new Beam transform follows the identical shape already in the repo: a
`Beam*Meta implements
IBeamPipelineTransformHandler` in `org.apache.hop.beam.transforms.<pkg>`, a
`PTransform` in
`org.apache.hop.beam.core.transform`, a `DoFn` in
`org.apache.hop.beam.core.fn`, i18n keys in
`messages_en_US.properties`, a test Meta, and a doc page registered in
`nav.adoc`.
---
## Effort ratings
| # | Issue | Effort | Wave |
|---|-------|--------|------|
| 5064 | BigQuery input RECORD (nested) fields | **Low** | 2 |
| 2416 | BigQuery `withQueryLocation` + `withoutValidation` | **Low** | 2 |
| 2337 | Beam File Output compression | **Low** | 2 |
| 2355 | `gcpTempLocation` on Dataflow run config | **Low** | 2 |
| 2675 | Kafka Avro + schema registry (2 defects) | **Medium** | 2 |
| 2040 | Beam partitioning support | **Medium** | 3 |
| 2379 | Add Sequence handler | **Medium** | 3 |
| 2275 | Window by key | **Medium** | 3 |
| 3131 | Get Flink JobID | **Low** (answer only) | 5 |
| 3478 | Partition / copies on Flink | **Low** (answer only) | 5 |
| 2326 | Beam plugin dev guidelines | **Medium** | 5 |
| 2357 | Parquet transform + ParquetIO | **Medium** | 3 |
| 2358 | Avro transform + AvroIO | **Medium** | 3 |
| 2375 | Beam MQTT IO | **Medium** | 4 |
| 2377 | Beam Splunk IO | **Medium** | 4 |
| 2378 | Beam Snowflake IO | **Medium** | 4 |
| 2409 | Beam Debezium IO | **High** | 4 |
| 2613 | Elastic/OpenSearch transforms + handler | **High** | 4 |
---
## Wave 0 — Triage table (do this first; it is the umbrella's actual
deliverable)
**T0.1** Create the triage record. It is the artefact that justifies every
later decision and
lets you drop sub-issues you should not build.
Write
`docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/issue-8708-triage.adoc`
with one
table row per sub-issue:
```adoc
////
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.
////
:description: Triage table for umbrella issue 8708 (Beam engine, runners and
Beam IO).
:page-marketplace: hop-engines-beam
= Issue 8708 triage table
[options="header",cols="1,1,1,4"]
|===
|Sub-issue |Effort |Decision |Rationale
|#2355|Low|Implement
|`GcpOptions.setGcpTempLocation` exists in Beam 2.74 and is already on the
classpath.
|#3131|Low|Close with answer
|`FlinkRunnerResult` discards the Flink `JobID`; no public accessor exists.
Document it.
|===
```
Fill in every row as you work the waves; do not pre-fill decisions you have
not verified.
**T0.2** Register the page in the nav, next to the other Beam entries
(`docs/hop-user-manual/modules/ROOT/nav.adoc`, after line 66):
```adoc
*** xref:pipeline/beam/issue-8708-triage.adoc[Beam issue triage (umbrella
#8708)]
```
**Verify**:
```bash
cd /home/matt/tmp/hop
grep -n "issue-8708-triage" docs/hop-user-manual/modules/ROOT/nav.adoc
# expected: one line, e.g. 67:***
xref:pipeline/beam/issue-8708-triage.adoc[Beam issue triage (umbrella #8708)]
```
**Commit**:
```bash
git add docs/hop-user-manual/modules/ROOT
git commit -m "Docs: add triage table for umbrella issue 8708"
```
---
## Wave 1 — Shared test scaffolding
**T1.1** Add a reusable base class for single-transform Beam pipeline tests
so waves 2–4 do not
each re-derive the `PipelineTestBase` boilerplate. Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/transform/SingleTransformPipelineTestBase.java`:
```java
/*
* 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 java.io.File;
import java.util.List;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.row.IRowMeta;
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.IEngineMetric;
import org.apache.hop.pipeline.transform.ITransformMeta;
import org.apache.hop.pipeline.transform.TransformMeta;
/**
* Base for pipeline tests that wire a single input transform, a single
transform under test and
* a single Beam Output transform, then assert on the produced rows. Extends
{@link
* PipelineTestBase} for plugin registration, metadata and the {@code
direct} run configuration.
*/
public abstract class SingleTransformPipelineTestBase extends
PipelineTestBase {
/** Runs the pipeline and returns the file contents of the single output
file. */
protected String runAndReadOutput(PipelineMeta pipelineMeta, String
outputFolder)
throws Exception {
File outputDir = new File(outputFolder);
if (outputDir.exists()) {
for (File file : Objects.requireNonNull(outputDir.listFiles())) {
if (!file.getName().startsWith(".")) {
file.delete();
}
}
} else {
outputDir.mkdirs();
}
createRunPipeline(variables, pipelineMeta);
File[] files = outputDir.listFiles((dir, name) -> !name.startsWith("."));
if (files == null || files.length == 0) {
throw new HopException("No output files were produced in " +
outputDir);
}
StringBuilder result = new StringBuilder();
for (File file : files) {
result.append(Objects.requireNonNull(new
String(Files.readAllBytes(file.toPath()))));
}
return result.toString();
}
/** The number of lines the given component wrote, or -1 when the metric
is absent. */
protected long writtenLines(EngineMetrics metrics, IEngineComponent
component) {
IEngineMetric metric = metrics.getComponentMetric(component,
org.apache.hop.pipeline.Pipeline.METRIC_WRITTEN);
return metric == null ? -1L : metric;
}
}
```
Also add the imports the file needs (`java.io.File`, `java.nio.file.Files`,
`java.util.Objects`,
`java.util.List`) and drop the unused `ITransformMeta` / `IRowMeta` /
`TransformMeta` imports —
Spotless will not remove them for you, but an unused import is a review
smell. The final
compiling form is:
```java
package org.apache.hop.beam.transform;
import java.io.File;
import java.nio.file.Files;
import java.util.Objects;
import org.apache.hop.core.exception.HopException;
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.IEngineMetric;
```
**Verify** (it must compile before any wave-2 test can extend it):
```bash
cd /home/matt/tmp/hop
./mvnw -o -q -pl plugins/engines/beam -am -DskipTests
-Dspotless.check.skip=true -Drat.skip=true install
# expected: exit 0, no output
```
**Commit**:
```bash
git add plugins/engines/beam/src/test
git commit -m "Beam tests: add single-transform pipeline test base"
```
---
## Wave 2 — The five cheap, high-confidence code changes
### T2.1 — #5064 BigQuery input: support RECORD (nested) fields · effort
**Low**
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/core/fn/BQSchemaAndRecordToHopFnTest.java`.
The bug is in the type-derivation loop of
`BQSchemaAndRecordToHopFn.apply()`: `AvroType.valueOf("RECORD")` throws
`IllegalArgumentException` → wrapped as `HopRuntimeException("Unable to
recognize data type
'RECORD'")`. The fix maps RECORD to `IValueMeta.TYPE_STRING` and
JSON-encodes the nested value.
```java
/*
* ... Apache License 2.0 header (copy verbatim from BeamBQInputMeta.java
lines 1-16) ...
*/
package org.apache.hop.beam.core.fn;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.hop.beam.core.HopRow;
import org.apache.hop.core.row.IValueMeta;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
/** #5064: BigQuery RECORD (nested / repeated) fields must not blow up the
type derivation. */
class BQSchemaAndRecordToHopFnTest {
private static final String NESTED_SCHEMA_JSON =
"{\"type\":\"record\",\"name\":\"Row\",\"fields\":["
+ "{\"name\":\"id\",\"type\":\"long\"},"
+
"{\"name\":\"address\",\"type\":{\"type\":\"record\",\"name\":\"Address\",\"fields\":["
+ "{\"name\":\"city\",\"type\":\"string\"},"
+ "{\"name\":\"zip\",\"type\":\"string\"}]},"
+
"{\"name\":\"tags\",\"type\":{\"type\":\"array\",\"items\":\"string\"}}"
+ "]}";
@BeforeAll
static void init() {
org.apache.hop.core.HopEnvironment.init();
}
@Test
void avroTypeEnumKnowsRecordMapsToString() {
assertEquals(IValueMeta.TYPE_STRING,
BQSchemaAndRecordToHopFn.AvroType.RECORD.getHopType());
assertEquals(IValueMeta.TYPE_STRING,
BQSchemaAndRecordToHopFn.AvroType.STRUCT.getHopType());
}
@Test
void nestedRecordBecomesAJsonString() {
Schema schema = new Schema.Parser().parse(NESTED_SCHEMA_JSON);
GenericRecord address =
new GenericData.Record(schema.getField("address").schema()).apply();
address.put("city", "Ghent");
address.put("zip", "9000");
GenericRecord row = new GenericData.Record(schema);
row.put("id", 1L);
row.put("address", address);
row.put("tags", java.util.List.of("a", "b"));
TableFieldSchema idField = new
TableFieldSchema().setName("id").setType("INTEGER");
TableFieldSchema addressField =
new TableFieldSchema()
.setName("address")
.setType("RECORD")
.setMode("NULLABLE")
.setSchema(
new TableSchema()
.setFields(
Arrays.asList(
new
TableFieldSchema().setName("city").setType("STRING"),
new
TableFieldSchema().setName("zip").setType("STRING"))));
TableFieldSchema tagsField = new
TableFieldSchema().setName("tags").setType("STRING").setMode("REPEATED");
TableSchema tableSchema = new
TableSchema().setFields(Arrays.asList(idField, addressField, tagsField));
SchemaAndRecord schemaAndRecord =
SchemaAndRecord.of(tableSchema, row,
java.time.Instant.ofEpochSecond(0));
String rowMetaJson =
"{\"row\":{\"meta\":[{\"type\":\"Integer\",\"name\":\"id\",\"length\":-1,\"precision\":-1},"
+
"{\"type\":\"String\",\"name\":\"address\",\"length\":-1,\"precision\":-1},"
+
"{\"type\":\"String\",\"name\":\"tags\",\"length\":-1,\"precision\":-1}]}}";
BQSchemaAndRecordToHopFn fn = new BQSchemaAndRecordToHopFn("test",
rowMetaJson);
HopRow hopRow = assertDoesNotThrow(() -> fn.apply(schemaAndRecord));
assertEquals(1L, hopRow.getRow()[0]);
assertEquals("{\"city\":\"Ghent\",\"zip\":\"9000\"}",
hopRow.getRow()[1]);
assertEquals("[\"a\",\"b\"]", hopRow.getRow()[2]);
}
}
```
> If `SchemaAndRecord.of(...)` has a different signature in Beam 2.74,
compile the test first and
> let javac tell you; do not guess. `javap` on the jar is the authority:
> `javap -cp
~/.m2/repository/org/apache/beam/beam-sdks-java-io-google-cloud-platform/2.74.0/beam-sdks-java-io-google-cloud-platform-2.74.0.jar
org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord`
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BQSchemaAndRecordToHopFnTest test 2>&1 |
grep -E "ERROR.*AvroType|No enum constant|BUILD"
# expected: a compilation error naming AvroType.RECORD / AvroType.STRUCT
(they do not exist yet)
```
**GREEN.** Two edits in
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/BQSchemaAndRecordToHopFn.java`.
1. In the `AvroType` enum (currently lines 179–192), add the two entries
before the trailing `;`:
```java
STRING(IValueMeta.TYPE_STRING),
BYTES(IValueMeta.TYPE_STRING),
INTEGER(IValueMeta.TYPE_INTEGER),
INT64(IValueMeta.TYPE_INTEGER),
FLOAT(IValueMeta.TYPE_NUMBER),
FLOAT64(IValueMeta.TYPE_NUMBER),
BOOLEAN(IValueMeta.TYPE_BOOLEAN),
BOOL(IValueMeta.TYPE_BOOLEAN),
TIMESTAMP(IValueMeta.TYPE_DATE),
DATE(IValueMeta.TYPE_DATE),
TIME(IValueMeta.TYPE_DATE),
DATETIME(IValueMeta.TYPE_DATE),
// #5064: BigQuery nested (RECORD) and repeated (REPEATED over a RECORD)
fields have no
// native Hop type. We hand them back as JSON strings rather than
failing the whole read.
RECORD(IValueMeta.TYPE_STRING),
STRUCT(IValueMeta.TYPE_STRING),
;
```
2. In the conversion loop (currently lines 122–161), add a case **before**
the `default:`
branch that JSON-encodes non-primitive Avro values:
```java
case IValueMeta.TYPE_STRING:
// #5064: a nested RECORD/REPEATED value arrives as a
GenericRecord or a
// List<GenericRecord>. Neither has a useful toString(), so
serialise it.
if (srcData instanceof GenericRecord || srcData instanceof
List) {
row[index] = HopJson.newMapper().writeValueAsString(srcData);
} else {
row[index] = srcData.toString();
}
break;
```
3. Add the import `org.apache.hop.core.json.HopJson` (it is already used
elsewhere in the repo,
e.g. `HopPipelineMetaToBeamPipelineConverter.java:52`).
4. Relax the "all fields covered" guard at lines 105–110 so that a field the
table schema does
not mention (common for `fromQuery` results) leaves a null rather than
throwing:
```java
// See that we got all the fields covered...
//
for (int i = 0; i < rowMeta.size(); i++) {
if (valueTypes[i] == 0) {
IValueMeta valueMeta = rowMeta.getValueMeta(i);
// #5064: only warn. A field we can't derive a type for (e.g. a
nested RECORD we
// were not told about) must not fail the whole read; the
conversion switch will
// fall back to the declared Hop type.
LOG.warn(
"Unable to determine the BigQuery type of field '"
+ valueMeta.getName()
+ "', falling back to the Hop type declared in the
transform");
valueTypes[i] = valueMeta.getType();
}
}
```
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BQSchemaAndRecordToHopFnTest test 2>&1 |
tail -12
# expected: "Tests run: 2, Failures: 0, Errors: 0" ... "BUILD SUCCESS"
```
**Also update the dialog.** `BeamBQInputDialog.getFields()` (line 271) does
a bare
`BQSchemaAndRecordToHopFn.AvroType.valueOf(type)` and will now succeed for
RECORD, but it drops
`newName`. Leave it — that is existing behaviour, out of scope.
**Docs.** Append a `RECORD / nested fields` note to
`docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/beambigqueryinput.adoc`.
**Commit**:
```bash
git add plugins/engines/beam/src docs/hop-user-manual
git commit -m "Fixes #5064: return BigQuery RECORD fields as JSON strings"
```
---
### T2.2 — #2416 BigQuery `withQueryLocation` + `withoutValidation` ·
effort **Low**
**RED.** Extend
`plugins/engines/beam/src/test/java/org/apache/hop/beam/transforms/bq/BeamBQInputMetaTest.java`
with two round-trip assertions, and add the two new elements to
`plugins/engines/beam/src/test/resources/beam-bq-input-transform.xml`:
```xml
<query_location>US</query_location>
<skip_query_validation>true</skip_query_validation>
```
```java
@Test
void xmlRoundTripPreservesQueryOptions() throws Exception {
BeamBQInputMeta meta =
TransformSerializationTestUtil.testSerialization(
"/beam-bq-input-transform.xml", BeamBQInputMeta.class);
assertEquals("US", meta.getQueryLocation());
assertTrue(meta.isSkippingQueryValidation());
}
```
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamBQInputMetaTest test 2>&1 | grep -E
"cannot find symbol|ERROR.*getQueryLocation|BUILD"
# expected: compilation failure — getQueryLocation() /
isSkippingQueryValidation() do not exist
```
**GREEN.**
1.
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/bq/BeamBQInputMeta.java`
—
add the two properties after the existing `query` property (line 64) and
**convert the class
to Lombok** for the accessors, per `AGENTS.md`. Concretely: add `@Getter
@Setter` on the
class, delete every hand-written `getX()`/`setX()` pair (lines 160–239),
and insert:
```java
@HopMetadataProperty(key = "query_location")
private String queryLocation;
@HopMetadataProperty(key = "skip_query_validation")
private boolean skippingQueryValidation;
```
Add `import lombok.Getter;` and `import lombok.Setter;`.
Pass both through to the transform (line 146–154):
```java
BeamBQInputTransform beamInputTransform =
new BeamBQInputTransform(
transformMeta.getName(),
transformMeta.getName(),
variables.resolve(projectId),
variables.resolve(datasetId),
variables.resolve(tableId),
variables.resolve(query),
variables.resolve(queryLocation),
skippingQueryValidation,
JsonRowMeta.toJson(outputRowMeta));
```
2.
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamBQInputTransform.java`
— new fields + constructor params (`queryLocation`,
`skippingQueryValidation`), and in
`expand()` replace the `if (StringUtils.isEmpty(query))` block (lines
93–97) with:
```java
if (StringUtils.isEmpty(query)) {
bqTypedRead =
BigQueryIO.read(toHopFn).from(tableReference).usingStandardSql();
} else {
bqTypedRead =
BigQueryIO.read(toHopFn).fromQuery(query).usingStandardSql();
// #2416: let the caller point at a US/EU query location and skip
validation when the
// query has not been registered yet.
if (StringUtils.isNotEmpty(queryLocation)) {
bqTypedRead = bqTypedRead.withQueryLocation(queryLocation);
}
if (skippingQueryValidation) {
bqTypedRead = bqTypedRead.withoutValidation();
}
}
```
3.
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/bq/BeamBQInputDialog.java`
— the dialog is hand-laid-out SWT (`ScrolledComposite` +
`FormAttachment`, lines 84–218). Per
`AGENTS.md` do **not** hand-lay-out new rows on the shell. Add the two
widgets instead as a
small `TableView`-free block placed *before* `wlQuery`, reusing the
existing project-id block
pattern verbatim (lines 103–118), with `GuiElementType`-equivalent widgets
(`TextVar` + `Button` for the checkbox). Register the getters/setters in
`getData()` and
`getInfo()`:
```java
in.setQueryLocation(wQueryLocation.getText());
in.setSkippingQueryValidation(wSkipQueryValidation.getSelection());
```
4. i18n — append to
`plugins/engines/beam/src/main/resources/org/apache/hop/beam/transforms/bq/messages/messages_en_US.properties`:
```properties
BeamBQInputDialog.QueryLocation=Query location
BeamBQInputDialog.QueryLocation.Tooltip=The BigQuery dataset location that
runs the query, for example 'EU' or 'US'. Leave blank to use the job location.
BeamBQInputDialog.SkipQueryValidation=Skip query validation
BeamBQInputDialog.SkipQueryValidation.Tooltip=Do not validate the query
before the job starts. Use this for a query which is created or changed by an
earlier step.
```
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamBQInputMetaTest test 2>&1 | tail -10
# expected: "Tests run: 6, Failures: 0, Errors: 0" (4 existing + 1 new + ...
) "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam
git commit -m "Fixes #2416: add BigQuery query location and skip-validation
options"
```
---
### T2.3 — #2337 Beam File Output compression · effort **Low**
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/core/transform/BeamOutputCompressionTest.java`
that asserts the meta round-trips a compression setting and that the
produced file is really
gzipped:
```java
/*
* ... Apache License 2.0 header ...
*/
package org.apache.hop.beam.core.transform;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.zip.GZIPInputStream;
import org.apache.beam.sdk.io.Compression;
import org.apache.hop.beam.transforms.io.BeamOutputMeta;
import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.plugins.PluginRegistry;
import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
/** #2337: the Beam file output transform must be able to write compressed
files. */
class BeamOutputCompressionTest {
@BeforeAll
static void initHopEnvironment() throws Exception {
HopEnvironment.init();
PluginRegistry.init();
}
@Test
void compressionRoundTripsThroughTheTransformXml() throws Exception {
BeamOutputMeta meta =
TransformSerializationTestUtil.testSerialization(
"/beam-output-compressed-transform.xml", BeamOutputMeta.class);
assertEquals("GZIP", meta.getCompression());
}
@Test
void compressionConstantIsSupported() {
// Guards the value we write into the XML: it has to be a real Beam
Compression constant.
assertEquals(Compression.GZIP, Compression.valueOf("GZIP"));
assertTrue(Compression.GZIP.isCompressed("out.gz"));
}
}
```
with fixture
`plugins/engines/beam/src/test/resources/beam-output-compressed-transform.xml`
(copy the header + skeleton of `beam-bq-input-transform.xml`,
`<type>BeamOutput</type>`, and):
```xml
<output_location>/tmp/customers/output</output_location>
<compression>GZIP</compression>
<windowed>N</windowed>
```
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamOutputCompressionTest test 2>&1 | grep
-E "cannot find symbol|getCompression|BUILD"
# expected: compilation failure — getCompression() does not exist on
BeamOutputMeta
```
**GREEN.**
1.
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/io/BeamOutputMeta.java`
— add `@HopMetadataProperty private String compression;` next to
`windowed` (line 69), plus
getter/setter, and pass it into the transform (lines 142–151):
```java
BeamOutputTransform beamOutputTransform =
new BeamOutputTransform(
transformMeta.getName(),
variables.resolve(outputLocation),
variables.resolve(filePrefix),
variables.resolve(fileSuffix),
variables.resolve(outputFileDefinition.getSeparator()),
variables.resolve(outputFileDefinition.getEnclosure()),
windowed,
variables.resolve(compression),
JsonRowMeta.toJson(rowMeta));
```
2.
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamOutputTransform.java`
— add the field, the constructor parameter, and in `expand()` after the
`withSuffix` block
(line 112):
```java
// #2337: TextIO can write compressed files. AUTO derives the codec
from the suffix, so
// it needs no extra configuration from the user.
if (StringUtils.isNotEmpty(compression)) {
try {
write = write.withCompression(Compression.valueOf(compression));
} catch (IllegalArgumentException e) {
throw new HopRuntimeException(
"Unknown compression '"
+ compression
+ "' for Beam output transform '"
+ transformName
+ "'. Use one of AUTO, UNCOMPRESSED, GZIP, BZIP2, ZIP,
ZSTD, LZO, LZOP, "
+ "DEFLATE or SNAPPY.",
e);
}
}
```
Add imports `org.apache.beam.sdk.io.Compression` and
`org.apache.hop.core.exception.HopRuntimeException`.
3. Dialog
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/io/BeamOutputDialog.java`
— add a `ComboVar`/combo populated from `Compression.values()` above the
`windowed` checkbox,
following the block pattern already used for `outputLocation`. Wire
`getData()`/`getInfo()`.
4. i18n — append to
`plugins/engines/beam/src/main/resources/org/apache/hop/beam/transforms/io/messages/messages_en_US.properties`:
```properties
BeamOutputDialog.Compression=Compression
BeamOutputDialog.Compression.Tooltip=How to compress the output files.
'AUTO' derives the codec from the file suffix, for example '.gz'.
```
5. Docs — add a Compression row to the Options table in
`docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/beamfileoutput.adoc`.
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamOutputCompressionTest test 2>&1 | tail
-8
# expected: "Tests run: 2, Failures: 0, Errors: 0" "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #2337: add compression support to the Beam file output
transform"
```
---
### T2.4 — #2355 `gcpTempLocation` on the Dataflow run configuration ·
effort **Low**
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/engines/dataflow/BeamDataFlowPipelineRunConfigurationTest.java`:
```java
/*
* ... Apache License 2.0 header ...
*/
package org.apache.hop.beam.engines.dataflow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
import org.apache.hop.core.HopEnvironment;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
/** #2355: BigQuery IO needs a GCS temp location, separate from the generic
temp location. */
class BeamDataFlowPipelineRunConfigurationTest {
@BeforeAll
static void initHopEnvironment() throws Exception {
HopEnvironment.init();
}
@Test
void gcpTempLocationIsPropagatedToThePipelineOptions() throws Exception {
BeamDataFlowPipelineRunConfiguration runConfiguration =
new BeamDataFlowPipelineRunConfiguration();
runConfiguration.setGcpTempLocation("gs://my-bucket/beam-temp");
DataflowPipelineOptions options =
(DataflowPipelineOptions) runConfiguration.getPipelineOptions();
assertEquals("gs://my-bucket/beam-temp",
options.as(GcpOptions.class).getGcpTempLocation());
}
@Test
void cloneCopiesTheGcpTempLocation() {
BeamDataFlowPipelineRunConfiguration original = new
BeamDataFlowPipelineRunConfiguration();
original.setGcpTempLocation("gs://my-bucket/beam-temp");
assertTrue(original.clone().getGcpTempLocation().contains("gs://my-bucket/beam-temp"));
}
}
```
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamDataFlowPipelineRunConfigurationTest
test 2>&1 | grep -E "cannot find symbol|setGcpTempLocation|BUILD"
# expected: compilation failure — setGcpTempLocation() does not exist
```
**GREEN.** In
`plugins/engines/beam/src/main/java/org/apache/hop/beam/engines/dataflow/BeamDataFlowPipelineRunConfiguration.java`:
1. New widget + property after `gcpStagingLocation` (line 81). Note the
neighbouring staging
location is `GuiElementType.FOLDER`; the temp location is a GCS path, not
a local folder, so
use `TEXT`:
```java
@GuiWidgetElement(
order = "20025-dataflow-options",
parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
type = GuiElementType.TEXT,
toolTip = "i18n::BeamEnginesDataflow.OptionsGcpTempLocation.ToolTip",
label = "i18n::BeamEnginesDataflow.OptionsGcpTempLocation.Label")
@HopMetadataProperty
private String gcpTempLocation;
```
2. Copy constructor (line 203 area): `this.gcpTempLocation =
config.gcpTempLocation;`
3. In `getPipelineOptions()`, right after the staging-location call (line
234):
```java
if (StringUtils.isNotEmpty(getGcpTempLocation())) {
// #2355: BigQuery load jobs and other GCP IOs need a GCS path in
gcpTempLocation, which is
// a separate option from the generic tempLocation.
options
.as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class)
.setGcpTempLocation(resolve(getGcpTempLocation()));
}
```
4. Getter/setter pair at the end of the class.
5. i18n — append to
`plugins/engines/beam/src/main/resources/org/apache/hop/beam/engines/dataflow/messages/messages_en_US.properties`:
```properties
BeamEnginesDataflow.OptionsGcpTempLocation.Label=GCP temp location
BeamEnginesDataflow.OptionsGcpTempLocation.ToolTip=Cloud Storage path used
by transforms which must stage their data in GCS, for example the BigQuery
transforms. \n Must be a valid Cloud Storage URL, beginning with gs:// \n This
is different from the general 'Temp location' option.
```
6. Docs — add a `GCP temp location` row to the Options table in
`docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/beam-dataflow-pipeline-engine.adoc`
(insert after the `Staging location` row, line 66).
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamDataFlowPipelineRunConfigurationTest
test 2>&1 | tail -8
# expected: "Tests run: 2, Failures: 0, Errors: 0" "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #2355: add a gcpTempLocation option to the Dataflow run
configuration"
```
---
### T2.5 — #2675 Kafka Avro with a schema registry · effort **Medium** (two
defects)
#### T2.5a — the Confluent jars are not staged into the fat jar
**Defect**: `ClassNotFoundException:
io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe` on
the Flink task manager. The classes are on the compile classpath
(transitively) but
`plugins/engines/beam/lib` has no `lib` folder at all, and
`BeamConst.findLibraryFilesToStage`
(used by `FatJarBuilder`) only globs jar files under the named plugin
folders.
**Fix**: declare the Confluent artifacts explicitly in
`plugins/engines/beam/pom.xml` (so the version is pinned and visible) and
copy them into
`plugins/engines/beam/lib` as part of the build.
Add, next to the other `org.apache.kafka:kafka-clients` dependency:
```xml
<!-- #2675: the Kafka Avro / schema-registry transforms need the
Confluent serializers at
RUNTIME on the workers, not just at compile time. They are
staged into the Hop fat
jar by FatJarBuilder, which globs plugins/engines/beam/lib. -->
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>${confluent.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-schema-registry-client</artifactId>
<version>${confluent.version}</version>
</dependency>
```
and add a property next to `<spark.version>3.5.8</spark.version>` in that
same pom:
```xml
<confluent.version>7.6.1</confluent.version>
```
(7.6.1 is the version already resolved on this checkout — confirm with
`tr ':' '\n' < /tmp/beam-cp.txt | grep kafka-avro-serializer`.)
Then add a `maven-dependency-plugin` execution to the **same pom** so the
jars are always in
`lib/`:
```xml
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>copy-confluent-serializers-to-lib</id>
<phase>package</phase>
<goals>
<goal>copy-dependencies</goal>
</goals>
<configuration>
<includeGroupIds>io.confluent</includeGroupIds>
<outputDirectory>${project.basedir}/lib</outputDirectory>
<overWriteReleases>true</overWriteReleases>
</configuration>
</execution>
</executions>
</plugin>
```
**Test first.** Add to the existing
`plugins/engines/beam/src/test/java/org/apache/hop/beam/gui/HopBeamGuiPluginNativeFatJarTest.java`
(or a new sibling `FatJarStagingTest`) an assertion that the Confluent
classes the Kafka Avro
transforms need are on the module classpath:
```java
@Test
void confluentSerializersAreOnTheClasspathForTheKafkaAvroTransforms()
throws Exception {
// #2675: these were missing at runtime on the Flink task manager.
for (String className :
new String[] {
"io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe",
"io.confluent.kafka.serializers.KafkaAvroSerializer",
"io.confluent.kafka.schemaregistry.client.SchemaRegistryClient"
}) {
assertNotNull(
Class.forName(className),
className + " must be on the Beam plugin classpath for Kafka Avro
with a schema registry");
}
}
```
**Verify**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=HopBeamGuiPluginNativeFatJarTest test 2>&1
| tail -8
ls plugins/engines/beam/lib | head
# expected: test passes; ls shows kafka-avro-serializer-7.6.1.jar and friends
```
**Commit**:
```bash
git add plugins/engines/beam/pom.xml plugins/engines/beam/src/test
git commit -m "Fixes #2675 (a): stage the Confluent Kafka serializers with
the Beam fat jar"
```
#### T2.5b — the produce-side NPE in `AvroCoder.of(schema)`
**Defect**: `NullPointerException` at
`AvroCoder$AvroDeterminismChecker.checkIndexedRecord` from
`BeamProduceMeta.handleTransform:149` — `ValueMetaAvroRecord.getSchema()`
returns `null` when the
Avro value meta was produced by a transform that did not carry a schema
(e.g. an Avro Decode
transform output, or a schema read lazily from a registry), and
`AvroCoder.of(null)` NPEs.
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/transforms/kafka/BeamProduceMetaTest.java`:
```java
/*
* ... Apache License 2.0 header ...
*/
package org.apache.hop.beam.transforms.kafka;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hop.core.HopException;
import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.plugins.PluginRegistry;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.core.row.value.ValueMetaAvroRecord;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
/** #2675: the Kafka produce transform must explain itself instead of
throwing a raw NPE. */
class BeamProduceMetaTest {
@BeforeAll
static void initHopEnvironment() throws Exception {
HopEnvironment.init();
PluginRegistry.init();
}
@Test
void avroMessageFieldWithoutASchemaFailsWithAClearMessage() {
BeamProduceMeta meta = new BeamProduceMeta();
meta.setMessageField("avro");
meta.setKeyField("key");
IRowMeta rowMeta = new RowMeta();
rowMeta.addValueMeta(new
org.apache.hop.core.row.value.ValueMetaString("key"));
rowMeta.addValueMeta(new ValueMetaAvroRecord("avro"));
HopException exception =
assertThrows(
HopException.class,
() ->
meta.checkAvroMessageSchema(rowMeta,
org.apache.hop.core.variables.Variables.getADefaultVariableSpace()));
assertTrue(
exception.getMessage().contains("schema"),
"the message must name the missing schema, was: " +
exception.getMessage());
}
}
```
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamProduceMetaTest test 2>&1 | grep -E
"cannot find symbol|checkAvroMessageSchema|BUILD"
# expected: compilation failure — checkAvroMessageSchema does not exist
```
**GREEN.** In
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/kafka/BeamProduceMeta.java`,
extract the coder registration into a method that validates first (replacing
lines 147–154):
```java
// Register a coder for the short time that KV<HopRow, GenericRecord>
exists in Beam
//
checkAvroMessageSchema(rowMeta, variables);
if (rowMeta.searchValueMeta(messageFieldName).getType() ==
IValueMeta.TYPE_AVRO) {
ValueMetaAvroRecord valueMetaAvroRecord =
(ValueMetaAvroRecord) rowMeta.searchValueMeta(messageFieldName);
Schema schema = valueMetaAvroRecord.getSchema();
if (schema == null) {
throw new HopException(
"The Avro field '"
+ messageFieldName
+ "' in Kafka produce transform '"
+ transformMeta.getName()
+ "' has no schema. Add an Avro schema to the field, or use
a transform which "
+ "reads the schema from a registry (such as an Avro Decode
step with a schema "
+ "URL) before this transform.");
}
AvroCoder<GenericRecord> coder = AvroCoder.of(schema);
pipeline.getCoderRegistry().registerCoderForClass(GenericRecord.class,
coder);
}
```
and add the package-private helper the test calls:
```java
/**
* Verify that an Avro message field actually carries a schema. Beam's
{@code AvroCoder.of(null)}
* throws a bare NullPointerException deep inside the determinism checker,
which is useless to
* the person editing the pipeline (#2675).
*/
void checkAvroMessageSchema(IRowMeta rowMeta, IVariables variables) throws
HopException {
String field = variables.resolve(getMessageField());
IValueMeta valueMeta = rowMeta == null ? null :
rowMeta.searchValueMeta(field);
if (valueMeta == null) {
throw new HopException("Error finding message/value field " + field +
" in the input rows");
}
if (valueMeta.getType() == IValueMeta.TYPE_AVRO
&& ((ValueMetaAvroRecord) valueMeta).getSchema() == null) {
throw new HopException(
"The Avro field '"
+ field
+ "' has no schema attached. Beam cannot serialise an Avro
value without one; "
+ "attach a schema to the field before this Kafka produce
transform.");
}
}
```
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamProduceMetaTest test 2>&1 | tail -8
# expected: "Tests run: 1, Failures: 0, Errors: 0" "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam
git commit -m "Fixes #2675 (b): fail with a clear message when a Kafka Avro
field has no schema"
```
---
## Wave 3 — Transform-handler work on the existing converter
### T3.1 — #2379 Add Sequence handler · effort **Medium**
This is the template for every handler in this wave. `AddSequenceMeta`
(plugin id **`Sequence`**,
`plugins/transforms/addsequence/.../AddSequenceMeta.java:47`) currently
falls through to
`BeamGenericTransformHandler`, which runs the whole Hop transform inside a
`ParDo` on every
worker — so every worker produces its own sequence starting at the same
number.
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/pipeline/handler/BeamAddSequenceTransformHandlerTest.java`:
```java
/*
* ... Apache License 2.0 header ...
*/
package org.apache.hop.beam.pipeline.handler;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hop.beam.pipeline.HopPipelineMetaToBeamPipelineConverter;
import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.plugins.PluginRegistry;
import org.apache.hop.pipeline.transforms.addsequence.AddSequenceMeta;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
/** #2379: the Add Sequence transform needs a dedicated Beam handler. */
class BeamAddSequenceTransformHandlerTest {
@BeforeAll
static void initHopEnvironment() throws Exception {
HopEnvironment.init();
PluginRegistry.init();
}
@Test
void handlerIsRegisteredForTheAddSequenceTransform() {
HopPipelineMetaToBeamPipelineConverter converter =
new HopPipelineMetaToBeamPipelineConverter() {
// The default constructor needs a live pipeline + metadata; we
only exercise
// addDefaultTransformHandlers() here.
};
converter.addDefaultTransformHandlers();
assertTrue(
HopPipelineMetaToBeamPipelineConverter.EXPLICIT_HANDLER_PLUGIN_IDS.contains("Sequence"),
"Sequence must be advertised as explicitly supported on Beam");
assertNotNull(converter.getTransformHandlers().get("Sequence"));
}
@Test
void engineReportsTheSequenceTransformAsSupported() {
assertEquals(
org.apache.hop.core.plugins.EngineCompatibility.supported().isSupported(),
new
org.apache.hop.beam.engines.BeamPipelineEngineSupportsTestHelper()
.supportsSequence());
}
@Test
void sequenceMetaExposesTheRangeHopNeeds() {
AddSequenceMeta meta = new AddSequenceMeta();
meta.setStartAt("100");
meta.setIncrementBy("5");
meta.setValueName("seq");
assertEquals("seq", meta.getValueName());
}
}
```
> If the anonymous `HopPipelineMetaToBeamPipelineConverter` subclass does
not compile (the class has
> no no-arg constructor), replace the first test with a direct assertion on
the static set only:
>
`assertTrue(HopPipelineMetaToBeamPipelineConverter.EXPLICIT_HANDLER_PLUGIN_IDS.contains("Sequence"))`
> and drop `converter.addDefaultTransformHandlers()`. The static set is the
contract that
> `BeamPipelineEngine.supports()` reads, so that assertion alone is
sufficient. Do the same for
> the second test if `EngineCompatibility` has no `isSupported()` accessor —
assert on
> `HopPipelineMetaToBeamPipelineConverter.EXPLICIT_HANDLER_PLUGIN_IDS` in
both.
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamAddSequenceTransformHandlerTest test
2>&1 | grep -E "AssertionFailed|expected|BUILD"
# expected: failure — "Sequence must be advertised as explicitly supported
on Beam"
```
**GREEN.**
1. Add the constant to
`plugins/engines/beam/src/main/java/org/apache/hop/beam/util/BeamConst.java`
(after line 46):
```java
public static final String STRING_ADD_SEQUENCE_PLUGIN_ID = "Sequence";
```
2. Create
`plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/handler/BeamAddSequenceTransformHandler.java`.
Model it on `BeamRowGeneratorTransformHandler` (same directory) — extend
`BeamBaseTransformHandler`, `isInput()` false, `isOutput()` false, and in
`handleTransform`:
```java
// Don't simply cast but serialize/de-serialize the metadata to prevent
classloader exceptions
//
AddSequenceMeta meta = new AddSequenceMeta();
loadTransformMetadata(meta, transformMeta, metadataProvider,
pipelineMeta);
long startAt = Const.toLong(variables.resolve(meta.getStartAt()), 0L);
long incrementBy =
Const.toLong(variables.resolve(meta.getIncrementBy()), 1L);
if (incrementBy <= 0) {
throw new HopException(
"The Add Sequence transform '"
+ transformMeta.getName()
+ "' needs a positive increment, not '"
+ meta.getIncrementBy()
+ "'");
}
// The database-backed and counter-backed modes of the transform cannot
be expressed in Beam:
// every worker would keep its own counter. Refuse loudly rather than
silently emitting
// duplicate values.
if (meta.isDatabaseUsed() || meta.isCounterUsed()) {
throw new HopException(
"The Add Sequence transform '"
+ transformMeta.getName()
+ "' uses a database sequence or a pipeline counter, neither
of which is "
+ "available on Beam. Switch it to the increment-by mode
(clear the 'use database' "
+ "and 'use counter' options) and set 'Start at' so every
worker generates a "
+ "disjoint range.");
}
long limit = -1L;
if (StringUtils.isNotEmpty(meta.getMaxValue())) {
limit = Const.toLong(variables.resolve(meta.getMaxValue()), -1L);
}
// One contiguous, increasing sequence per copy of the transform.
GenerateSequence produces
// longs, so the ranges are cheap and the values are globally unique as
long as the
// start/copy arithmetic below adds up.
int copies = Math.max(1, transformMeta.getCopies(variables));
PCollection<Long> generated =
pipeline.apply(
transformMeta.getName(),
GenerateSequence.from(startAt).to(limit < 0 ? Long.MAX_VALUE :
limit));
PCollection<HopRow> out =
generated.apply(
ParDo.of(
new AddSequenceFn(
transformMeta.getName(),
JsonRowMeta.toJson(rowMeta),
variables.resolve(meta.getValueName()),
incrementBy,
copies)));
transformCollectionMap.put(transformMeta.getName(), out);
log.logBasic("Handled transform (ADD SEQUENCE) : " +
transformMeta.getName());
```
3. Create
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/AddSequenceFn.java`
— a
`DoFn<Long, HopRow>` that appends one `Integer` field named `valueName` to
each incoming row and
scales the value per copy so ranges stay disjoint:
```java
package org.apache.hop.beam.core.fn;
import java.io.Serial;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.hop.beam.core.HopRow;
import org.apache.hop.beam.core.BeamHop;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.JsonRowMeta;
import org.apache.hop.core.row.value.ValueMetaInteger;
import org.apache.hop.pipeline.Pipeline;
/** #2379: turns a Beam sequence of longs into a Hop row with one extra
Integer field. */
public class AddSequenceFn extends DoFn<Long, HopRow> {
@Serial private static final long serialVersionUID = 1L;
private final String transformName;
private final String rowMetaJson;
private final String valueName;
private final long incrementBy;
private final int copies;
private final Counter numErrors = Metrics.counter("main",
"BeamAddSequenceErrors");
private final Counter inputCounter =
Metrics.counter(Pipeline.METRIC_NAME_INPUT, transformName);
private final Counter writtenCounter =
Metrics.counter(Pipeline.METRIC_NAME_WRITTEN, transformName);
private transient IRowMeta rowMeta;
private transient IValueMeta valueMeta;
private transient int valueIndex;
public AddSequenceFn(
String transformName, String rowMetaJson, String valueName, long
incrementBy, int copies) {
this.transformName = transformName;
this.rowMetaJson = rowMetaJson;
this.valueName = valueName;
this.incrementBy = incrementBy;
this.copies = Math.max(1, copies);
}
@Setup
public void setUp() {
BeamHop.init();
rowMeta = JsonRowMeta.fromJson(rowMetaJson);
valueMeta = new ValueMetaInteger(valueName);
valueMeta.setOrigin(transformName);
rowMeta.addValueMeta(valueMeta);
valueIndex = rowMeta.size() - 1;
Metrics.counter(Pipeline.METRIC_NAME_INIT, transformName).inc();
}
@ProcessElement
public void processElement(ProcessContext context) {
try {
inputCounter.inc();
Long element = context.element();
Object[] row = new Object[valueIndex + 1];
long base = element * incrementBy;
row[valueIndex] = base;
context.output(new HopRow(row));
writtenCounter.inc();
} catch (Exception e) {
numErrors.inc();
throw new RuntimeException("Error adding a sequence value in " +
transformName, e);
}
}
}
```
> If the incoming `PCollection` already carries `HopRow`s (it does, when Add
Sequence sits
> downstream of a Beam input), the `DoFn` input type must be `HopRow`, not
`Long` — split the
> pipeline with `ParDo.of(new MapToSequenceFn(...))` before this one, or
simply make `AddSequenceFn`
> a `DoFn<HopRow, HopRow>` that passes the existing row through and appends
the field. Choose the
> `DoFn<HopRow, HopRow>` shape; it is the one that composes with the rest of
the converter, whose
> collections are all `PCollection<HopRow>`. The test in T3.1 does not
exercise this, so verify it
> with the end-to-end test in T3.2.
4. Register the handler — in
`plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/HopPipelineMetaToBeamPipelineConverter.java`:
```java
public static final Set<String> EXPLICIT_HANDLER_PLUGIN_IDS =
Set.of(
BeamConst.STRING_MERGE_JOIN_PLUGIN_ID,
BeamConst.STRING_BEAM_ROW_GENERATOR_PLUGIN_ID,
BeamConst.STRING_MEMORY_GROUP_BY_PLUGIN_ID,
BeamConst.STRING_ADD_SEQUENCE_PLUGIN_ID);
```
and in `addDefaultTransformHandlers()` (line 245 area):
```java
transformHandlers.put(
BeamConst.STRING_ADD_SEQUENCE_PLUGIN_ID, new
BeamAddSequenceTransformHandler());
```
5. Add the provided dependency to `plugins/engines/beam/pom.xml`, matching
the existing pattern:
```xml
<dependency>
<groupId>org.apache.hop</groupId>
<artifactId>hop-transform-addsequence</artifactId>
<scope>provided</scope>
</dependency>
```
6. i18n — no new dialog, so no new keys. But the transform icon/keywords:
nothing to do.
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamAddSequenceTransformHandlerTest test
2>&1 | tail -8
# expected: "Tests run: 3, Failures: 0, Errors: 0" "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam
git commit -m "Fixes #2379: add a Beam handler for the Add Sequence
transform"
```
### T3.2 — #2379 end-to-end test · effort **Low**
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/transform/AddSequencePipelineTest.java`,
extending `SingleTransformPipelineTestBase` (from T1.1) and modelled
line-for-line on
`plugins/engines/beam/src/test/java/org/apache/hop/beam/transform/GroupByPipelineTest.java`:
register `AddSequenceMeta` in `PipelineTestBase.setUp`'s `beamTransforms`
list, build a pipeline
Beam Input → Add Sequence → Beam Output, run it, and assert that the output
file contains
strictly increasing integers with no duplicates.
**Verify RED**: run it — it fails because the values repeat across workers
(`BeamGenericTransformHandler` gives every copy its own counter).
**Verify GREEN**: it passes after T3.1.
**Commit**: same commit as T3.1 or a follow-up — your call, but not a third
commit for one feature.
---
### T3.3 — #2275 Window by key · effort **Medium**
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/transforms/window/BeamWindowMetaTest.java`:
```java
/*
* ... Apache License 2.0 header ...
*/
package org.apache.hop.beam.transforms.window;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.plugins.PluginRegistry;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
/** #2275: the Beam window transform must support windowing on a key. */
class BeamWindowMetaTest {
@BeforeAll
static void initHopEnvironment() throws Exception {
HopEnvironment.init();
PluginRegistry.init();
}
@Test
void keyFieldRoundTripsThroughTheTransformXml() throws Exception {
BeamWindowMeta meta =
org.apache.hop.pipeline.transform.TransformSerializationTestUtil.testSerialization(
"/beam-window-keyed-transform.xml", BeamWindowMeta.class);
assertEquals("customerId", meta.getKeyField());
}
@Test
void defaultsAreUnkeyed() {
assertTrue(new BeamWindowMeta().getKeyField() == null || new
BeamWindowMeta().getKeyField().isEmpty());
}
}
```
with fixture
`plugins/engines/beam/src/test/resources/beam-window-keyed-transform.xml`
(`<type>BeamWindow</type>`, `<window_type>FIXED</window_type>`,
`<duration>60</duration>`,
`<key_field>customerId</key_field>`).
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamWindowMetaTest test 2>&1 | grep -E
"cannot find symbol|getKeyField|BUILD"
# expected: compilation failure — getKeyField() does not exist on
BeamWindowMeta
```
**GREEN.**
1.
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/window/BeamWindowMeta.java`:
add `@HopMetadataProperty(key = "key_field") private String keyField;`
(after line 90) with
getter/setter; clear it in `setDefault()`; and in `handleTransform`,
immediately after
`transformPCollection = input.apply(window);` (line 267), insert:
```java
// #2275: window per key instead of globally. Beam needs KV<K, V>, so
key the collection
// and re-key it afterwards; the key is a field on the Hop row, not a
separate output.
//
if (StringUtils.isNotEmpty(keyField)) {
String realKeyField = variables.resolve(keyField);
int keyIndex = rowMeta.indexOfValue(realKeyField);
if (keyIndex < 0) {
throw new HopException(
"Unable to find the key field '"
+ realKeyField
+ "' in the input of Beam Window transform '"
+ transformMeta.getName()
+ "'");
}
transformPCollection =
transformPCollection
.apply(WithKeys.of(new HopKeyFn(transformMeta.getName(),
rowMetaJson, keyIndex)))
.setCoder(KvCoder.of(StringUtf8Coder.of(),
transformPCollection.getCoder()))
.apply(GroupByKey.create())
.apply(Values.create())
.apply(Flatten.iterables());
}
```
Add imports for `WithKeys`, `GroupByKey`, `Values`, `Flatten`, `KvCoder`,
`StringUtf8Coder`.
2. Create
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/HopKeyFn.java`
— a
`DoFn<HopRow, KV<String, HopRow>>` that stringifies the key field:
```java
package org.apache.hop.beam.core.fn;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.values.KV;
import org.apache.hop.beam.core.BeamHop;
import org.apache.hop.beam.core.HopRow;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.JsonRowMeta;
import org.apache.hop.pipeline.Pipeline;
/** #2275: keys a HopRow on one of its fields so Beam can group it. */
public class HopKeyFn extends DoFn<HopRow, KV<String, HopRow>> {
private final String transformName;
private final String rowMetaJson;
private final int keyIndex;
private final Counter numErrors = Metrics.counter("main",
"BeamWindowKeyErrors");
private final Counter inputCounter =
Metrics.counter(Pipeline.METRIC_NAME_INPUT, transformName);
private final Counter writtenCounter =
Metrics.counter(Pipeline.METRIC_NAME_WRITTEN, transformName);
private transient IRowMeta rowMeta;
private transient IValueMeta keyValueMeta;
public HopKeyFn(String transformName, String rowMetaJson, int keyIndex) {
this.transformName = transformName;
this.rowMetaJson = rowMetaJson;
this.keyIndex = keyIndex;
}
@Setup
public void setUp() {
BeamHop.init();
rowMeta = JsonRowMeta.fromJson(rowMetaJson);
keyValueMeta = rowMeta.getValueMeta(keyIndex);
}
@ProcessElement
public void processElement(ProcessContext context) {
try {
inputCounter.inc();
HopRow row = context.element();
String key = keyValueMeta.getString(row.getRow());
if (key == null) {
key = "";
}
context.output(KV.of(key, row));
writtenCounter.inc();
} catch (Exception e) {
numErrors.inc();
throw new RuntimeException("Error keying rows in " + transformName, e);
}
}
}
```
3. Dialog `BeamWindowDialog.java` — add a text widget for the key field,
placed with the other
window fields. This dialog is hand-laid-out; follow its existing block
pattern rather than
inventing a new layout.
4. i18n — append to
`plugins/engines/beam/src/main/resources/org/apache/hop/beam/transforms/window/messages/messages_en_US.properties`:
```properties
BeamWindowDialog.KeyField=Window on this key
BeamWindowDialog.KeyField.Tooltip=Optional field to window on. When set,
windows are computed per value of this field instead of across the whole
stream. Leave blank to window globally.
```
5. Docs — add the key field to
`docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/beamwindow.adoc`.
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamWindowMetaTest test 2>&1 | tail -8
# expected: "Tests run: 2, Failures: 0, Errors: 0" "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #2275: let the Beam window transform window on a key"
```
---
### T3.4 — #2040 Beam partitioning support · effort **Medium**
Hop's own partitioning (`TransformMeta.isPartitioned()`,
`engine/.../TransformMeta.java:598`) is
currently meaningless on Beam: the converter never reads it, so a
partitioned transform sees all
the data on every worker.
**RED.** Create
`plugins/engines/beam/src/test/java/org/apache/hop/beam/core/partition/FieldHashPartitionFnTest.java`:
```java
/*
* ... Apache License 2.0 header ...
*/
package org.apache.hop.beam.core.partition;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hop.beam.core.HopRow;
import org.apache.hop.core.HopEnvironment;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
/** #2040: partitioning must be a stable, field-based hash, not "everything
to partition 0". */
class FieldHashPartitionFnTest {
@BeforeAll
static void initHopEnvironment() throws Exception {
HopEnvironment.init();
}
@Test
void sameKeyAlwaysLandsInTheSamePartition() {
String rowMetaJson =
"{\"row\":{\"meta\":[{\"type\":\"String\",\"name\":\"customerId\",\"length\":-1,\"precision\":-1}]}}";
FieldHashPartitionFn fn = new FieldHashPartitionFn("test", rowMetaJson,
0);
HopRow first = new HopRow(new Object[] {"CUST-1"});
HopRow second = new HopRow(new Object[] {"CUST-1"});
HopRow other = new HopRow(new Object[] {"CUST-2"});
int a = fn.partitionFor(first, 8);
int b = fn.partitionFor(second, 8);
int c = fn.partitionFor(other, 8);
assertEquals(a, b, "the same key must always hash to the same
partition");
assertTrue(a >= 0 && a < 8);
assertTrue(c >= 0 && c < 8);
}
@Test
void partitionIsNeverOutOfRange() {
String rowMetaJson =
"{\"row\":{\"meta\":[{\"type\":\"String\",\"name\":\"customerId\",\"length\":-1,\"precision\":-1}]}}";
FieldHashPartitionFn fn = new FieldHashPartitionFn("test", rowMetaJson,
0);
for (int i = 0; i < 100; i++) {
int p = fn.partitionFor(new HopRow(new Object[] {"CUST-" + i}), 4);
assertTrue(p >= 0 && p < 4, "partition " + p + " out of range for 4
partitions");
}
}
}
```
**Verify RED**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=FieldHashPartitionFnTest test 2>&1 | grep
-E "cannot find symbol|FieldHashPartitionFn|BUILD"
# expected: compilation failure — the class does not exist
```
**GREEN.**
1. Create
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/partition/FieldHashPartitionFn.java`:
```java
package org.apache.hop.beam.core.partition;
import java.io.Serial;
import java.util.Objects;
import org.apache.beam.sdk.transforms.Partition;
import org.apache.hop.beam.core.BeamHop;
import org.apache.hop.beam.core.HopRow;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.JsonRowMeta;
/**
* #2040: partitions a HopRow on the value of one field, using a stable
hash. The existing {@link
* SinglePartitionFn} sends every row to partition 0, which serialises the
whole pipeline.
*/
public class FieldHashPartitionFn implements Partition.PartitionFn<HopRow> {
@Serial private static final long serialVersionUID = 1L;
private final String rowMetaJson;
private final int keyIndex;
private transient IRowMeta rowMeta;
private transient IValueMeta keyValueMeta;
public FieldHashPartitionFn(String rowMetaJson, int keyIndex) {
this.rowMetaJson = rowMetaJson;
this.keyIndex = keyIndex;
}
@Override
public int partitionFor(HopRow elem, int numPartitions) {
if (numPartitions <= 1) {
return 0;
}
if (rowMeta == null) {
// initialize() is called by Beam on the worker; the field access
below is lazily inited
// for the direct-runner case where the fn is also exercised from a
unit test.
BeamHop.init();
rowMeta = JsonRowMeta.fromJson(rowMetaJson);
keyValueMeta = rowMeta.getValueMeta(keyIndex);
}
Object value = elem == null ? null : elem.getRow()[keyIndex];
String key = value == null ? "" : String.valueOf(value);
return Math.floorMod(Objects.hash(key), numPartitions);
}
}
```
(The test above calls a 3-arg constructor `("test", rowMetaJson, 0)`;
either give the class
that 3-arg form with an unused `transformName` for metrics, or change the
test to the 2-arg form
`new FieldHashPartitionFn(rowMetaJson, 0)`. Pick one and make both agree
— prefer the 2-arg form
and update the test.)
2. Wire it into the converter. In
`plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/HopPipelineMetaToBeamPipelineConverter.java`,
in `handleGenericTransform`, right after the `input` PCollection is resolved
and **before** the
handler is invoked (around line 625):
```java
// #2040: honour Hop's partitioning so keys that belong together
stay on one worker.
//
if (transformMeta.isPartitioned() && input != null) {
String partitionField = getSinglePartitionField(variables,
transformMeta);
if (StringUtils.isNotEmpty(partitionField)) {
int keyIndex = rowMeta.indexOfValue(partitionField);
if (keyIndex >= 0) {
input =
input.apply(
Partition.of(
new
FieldHashPartitionFn(JsonRowMeta.toJson(rowMeta), keyIndex)));
} else {
log.logBasic(
CONST_TRANSFORM
+ transformMeta.getName()
+ " is partitioned on '"
+ partitionField
+ "' which is not a field of its input; the
partitioning is ignored");
}
}
}
```
with the helper:
```java
/** The single field Hop's partitioning UI allows, or null when it is not
a plain field. */
private String getSinglePartitionField(IVariables variables, TransformMeta
transformMeta) {
List<String> fields =
transformMeta.getTransformPartitioningMeta() == null
? null
:
transformMeta.getTransformPartitioningMeta().getPartitionFields();
if (fields == null || fields.size() != 1) {
return null;
}
return variables.resolve(fields.get(0));
}
```
`getTransformPartitioningMeta()` / `getPartitionFields()` are the real
accessors on
`TransformMeta`; confirm them with
`grep -n "PartitioningMeta"
engine/src/main/java/org/apache/hop/pipeline/transform/TransformMeta.java`
before writing this and adapt. Add `import
org.apache.beam.sdk.transforms.Partition;`.
3. Docs — explain the semantics (and the `SinglePartitionFn` history) in
`docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/getting-started-with-beam.adoc`.
**Verify GREEN**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=FieldHashPartitionFnTest test 2>&1 | tail
-8
# expected: "Tests run: 2, Failures: 0, Errors: 0" "BUILD SUCCESS"
```
**Commit**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #2040: honour Hop partitioning in the Beam pipeline
converter"
```
---
### T3.5 — #2357 Parquet transform + ParquetIO · effort **Medium**
`ParquetIO` is **not** on the classpath; it lives in
`org.apache.beam:beam-sdks-java-io-parquet`.
`ParquetFileInput` / `ParquetFileOutput` metas are at
`plugins/tech/parquet/src/main/java/org/apache/hop/parquet/transforms/{input,output}/`.
1. Add to `plugins/engines/beam/pom.xml`:
```xml
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parquet</artifactId>
</dependency>
<dependency>
<groupId>org.apache.hop</groupId>
<artifactId>hop-transform-parquet</artifactId>
<scope>provided</scope>
</dependency>
```
2. Create
`plugins/engines/beam/src/main/java/org/apache/hop/beam/pipeline/handler/BeamParquetTransformHandler.java`
(one class handling both directions, `isInput()`/`isOutput()` driven by
whether the meta is
`ParquetFileInputMeta` or `ParquetOutputMeta`), using
`ParquetIO.sink(schema)` /
`ParquetIO.read(schema)` and a `ParquetFn`-style `DoFn` in
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/fn/`.
3. Add the two plugin ids to `EXPLICIT_HANDLER_PLUGIN_IDS` and
`addDefaultTransformHandlers()`.
4. Tests: `BeamParquetTransformHandlerTest` (registration + id assertions)
plus an end-to-end
Parquet write→read pipeline test extending
`SingleTransformPipelineTestBase`, using the
Direct runner so no cluster is needed.
5. Docs:
`docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/parquetfileoutput.adoc`
and `parquetfileinput.adoc` get a "On Beam" section; register in
`nav.adoc`.
**Verify**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=BeamParquetTransformHandlerTest test 2>&1
| tail -8
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest test 2>&1 | tail -8
# expected: BUILD SUCCESS twice
```
**Commit**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #2357: add a Beam Parquet handler for the Parquet file
transforms"
```
---
### T3.6 — #2358 Avro transform + AvroIO · effort **Medium**
`org.apache.beam.sdk.extensions.avro.io.AvroIO` **is** already on the
classpath (confirmed via
`dependency:build-classpath`), so this one is mostly handler code.
1. `plugins/engines/beam/pom.xml`: add only the Hop-side provided deps —
`hop-transform-avro` (if that artifactId exists; confirm with
`ls plugins/tech/avro && grep -n artifactId plugins/tech/avro/pom.xml`).
2. Create `BeamAvroTransformHandler` + `AvroSourceFn` / `AvroSinkFn` under
`org.apache.hop.beam.pipeline.handler` and `org.apache.hop.beam.core.fn`,
handling
`AvroFileInputMeta` (`id = "AvroFileInput"`) and `AvroOutputMeta` (`id =
"AvroOutput"`).
3. Register both ids in `EXPLICIT_HANDLER_PLUGIN_IDS` /
`addDefaultTransformHandlers()`.
4. Tests + docs, same shape as T3.5.
**Verify**: same two commands as T3.5.
**Commit**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #2358: add a Beam Avro handler for the Avro file
transforms"
```
---
## Wave 4 — The five new Beam IO transforms
All five follow **exactly** the shape of an existing IO transform. Use
`plugins/engines/beam/src/main/java/org/apache/hop/beam/transforms/pubsub/BeamPublish*`
(Meta / Data / Dialog) plus
`plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamPublishTransform.java`
as the copy-paste template — it is the smallest complete input+output pair
in the module.
Each of T4.1–T4.5 gets:
1. **pom**: add the Beam IO artifact (all confirmed present at `2.74.0`):
```xml
<dependency><groupId>org.apache.beam</groupId><artifactId>beam-sdks-java-io-mqtt</artifactId></dependency>
```
plus the matching Hop transform plugin as `provided` where one exists.
2. **`.../transforms/<pkg>/Beam<Name>Meta.java`** — `@Transform(id =
"Beam<Name>", image = "beam-<name>.svg",
..., supportedEngines = {"Beam*"})`, `extends BaseTransformMeta<...>
implements IBeamPipelineTransformHandler`,
Lombok `@Getter @Setter`, all properties `@HopMetadataProperty`.
3. **`.../core/transform/Beam<Name>Transform.java`** — the `PTransform`
wrapping the Beam IO class.
4. **`.../transforms/<pkg>/Beam<Name>Dialog.java`** — SWT dialog; follow the
`BeamPublishDialog`
block pattern, do not flatten the form.
5. **i18n** `.../transforms/<pkg>/messages/messages_en_US.properties`
(Apache header first).
6. **Icon** `plugins/engines/beam/src/main/resources/beam-<name>.svg` + the
two transform icon
copies under
`docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/`.
7. **Tests** `Beam<Name>MetaTest` (XML round-trip via
`TransformSerializationTestUtil`) and a
Direct-runner end-to-end test where one is feasible without a live broker.
8. **Docs**
`docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/beam<name>.adoc` +
a `nav.adoc` entry.
| Task | Issue | Artifact to add | Beam class to wrap | Effort |
|------|-------|-----------------|--------------------|--------|
| T4.1 | #2375 | `beam-sdks-java-io-mqtt` |
`org.apache.beam.sdk.io.mqtt.MqttIO` | Medium |
| T4.2 | #2377 | `beam-sdks-java-io-splunk` |
`org.apache.beam.sdk.io.splunk.SplunkIO` | Medium |
| T4.3 | #2378 | `beam-sdks-java-io-snowflake` |
`org.apache.beam.sdk.io.snowflake.SnowflakeIO` | Medium |
| T4.4 | #2409 | `beam-sdks-java-io-debezium` |
`org.apache.beam.sdk.io.debezium.DebeziumIO` | **High** — unbounded source
only, so it must be an input transform, and it needs its own run-configuration
notes |
| T4.5 | #2613 | `beam-sdks-java-io-elasticsearch` |
`org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO` | **High** — the issue
also asks for a Hop-side *Elastic* transform, not just a Beam handler; scope
that first |
**Before T4.5, settle the scope with the maintainer** (see Open questions).
If the Hop-side
Elastic transform is out of scope, land only the Beam handler for the
existing transform and say so
in the closing comment.
**Verify, per task**:
```bash
cd /home/matt/tmp/hop
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest -Dtest=Beam<Name>MetaTest test 2>&1 | tail -8
# expected: BUILD SUCCESS
```
**Commit, per task**:
```bash
git add plugins/engines/beam docs/hop-user-manual
git commit -m "Fixes #<issue>: add the Beam <Name> IO transform"
```
---
## Wave 5 — Documentation and the answers-only issues
### T5.1 — #2326 Beam plugin development guidelines · effort **Medium**
Create
`docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/developing-beam-transforms.adoc`.
It must answer the three questions the issue actually asks, with real class
names from this repo:
1. **How does a Hop transform become a Beam step?** Top-down:
`BeamPipelineEngine.prepareExecution()` →
`HopPipelineMetaToBeamPipelineConverter.createPipeline()`
→ `handleBeamInputTransforms()` / `handleGenericTransform()` /
`handleBeamOutputTransforms()` →
the resolved `IBeamPipelineTransformHandler.handleTransform(...)`.
2. **How do I write one?** Three options, in order of preference:
- Make the meta implement `IBeamPipelineTransformHandler` (see
`BeamBQInputMeta.java:52`) and build a `PTransform` yourself.
- Implement `IBeamPipelineTransformHandler` on a standalone handler class
and register it in
`HopPipelineMetaToBeamPipelineConverter.addDefaultTransformHandlers()`
+ the
`EXPLICIT_HANDLER_PLUGIN_IDS` set (see
`BeamRowGeneratorTransformHandler`).
- Do nothing: `BeamGenericTransformHandler` wraps the transform in a
`ParDo` and it usually
just works — until it silently does not.
3. **What are the Meta / Data / Dialog classes?**
`BaseTransformMeta<Transform, TransformData>`,
the `Transform` (runtime) class, and the SWT `BaseTransformDialog`;
annotate properties with
`@HopMetadataProperty` and widgets with `@GuiWidgetElement`, keep them in
a `group` /
`groupType` so `GuiCompositeWidgets` can lay them out (see `AGENTS.md`).
Also document: the metrics contract
(`Pipeline.METRIC_NAME_INPUT/WRITTEN/ERROR/INIT` counters
named after the transform, which is what
`BeamPipelineEngine.populateEngineMetrics()` reads), the
`HARD_BANNED_META_TYPES` list, and the fact that every `handleTransform`
implementation must call
`BeamHop.init()` on the worker.
Register in `nav.adoc` under the Beam block.
**Verify**:
```bash
cd /home/matt/tmp/hop
grep -n "developing-beam-transforms"
docs/hop-user-manual/modules/ROOT/nav.adoc
grep -c "IBeamPipelineTransformHandler"
docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/developing-beam-transforms.adoc
# expected: 1 nav hit; >= 6 mentions of the interface
```
**Commit**:
```bash
git add docs/hop-user-manual
git commit -m "Fixes #2326: document how to develop Apache Beam transforms
in Hop"
```
### T5.2 — #3131 Get the Flink JobID · effort **Low** (answer only)
Do **not** write code. Post a comment on #3131 with the evidence and the
workaround, then close
it as answered:
> Apache Beam 2.74 does not surface the Flink `JobID`. `FlinkRunnerResult`
is constructed from
> `JobExecutionResult` in
`FlinkPipelineExecutionEnvironment.createAttachedPipelineResult()` and
> keeps only the accumulators and the runtime; the detached variant
`FlinkDetachedRunnerResult`
> holds a `JobClient` in a private field with no accessor. There is no
supported way to read the
> JobID from `PipelineResult`.
>
> What to use instead: the Beam job name, which Hop already derives from the
pipeline name
> (`HopPipelineMetaToBeamPipelineConverter.sanitizeJobName()` sets
`pipelineOptions.setJobName(...)`).
> On a Flink cluster that maps to the Flink job name, and on a Kubernetes
deployment it appears in
> the Flink Web UI under "Jobs" / "Running Jobs" — match on the name rather
than the ID.
>
> For Dataflow, the JobID *is* available and Hop already reports it
> (`BeamDataFlowPipelineEngine.DETAIL_DATAFLOW_JOB_ID`).
Then tick the box on #8708.
### T5.3 — #3478 Partition and copies on Flink · effort **Low** (answer
only)
Post the answer, referencing what this branch changed, and close:
> On a Beam runner, "copies" and "distribute" describe the *local* Hop
engine, not the remote one:
> a Beam cluster decides its own parallelism from the runner and the data.
Hop's partitioning is
> now honoured on Beam by keying the rows on the configured field so a key
stays on one worker
> (see #2040); setting more copies than one does **not** create more Beam
workers.
>
> For Flink, set `Parallelism` in the Flink pipeline run configuration
> (`BeamFlinkPipelineRunConfiguration.flinkParallelism`), and force a single
thread for a specific
> transform with the `SingleThreaded` attribute under the `Hop Beam` group
> (`BeamGenericTransformHandler.needsSingleThreading()`), which reduces the
parallelism of that
> step to one.
### T5.4 — Close out the umbrella · effort **Low**
1. Finish every row of the triage table from T0.1.
2. Verify the build is clean end to end:
```bash
cd /home/matt/tmp/hop
./mvnw spotless:apply
./mvnw -o -pl plugins/engines/beam -DskipTests -Drat.skip=true install
2>&1 | tail -5
./mvnw -o -pl plugins/engines/beam -Dspotless.check.skip=true
-Drat.skip=true -Pskip-uitest test 2>&1 | tail -10
# expected: BUILD SUCCESS, and "Tests run: N, Failures: 0, Errors: 0"
with N > 67
git status --short
# expected: clean, or only the plan file untracked
```
3. Comment on #8708 with a table of every sub-issue → outcome → commit sha,
and tick the boxes.
4. Close #8708 once every sub-issue is either closed or explicitly declined.
---
## Tests / validation (summary)
- **Per task**: write the failing test → run it and *see it fail* →
implement → run and see it pass
→ `./mvnw spotless:apply` → commit. The RED step is not optional; several
of these defects
(the RECORD crash, the Avro coder NPE) only reproduce as a specific
exception, and a test that
never failed proves nothing.
- **Per wave**: `./mvnw -o -pl plugins/engines/beam
-Dspotless.check.skip=true -Drat.skip=true -Pskip-uitest test`
must stay at `Failures: 0, Errors: 0` and the test count must only go up
(baseline: 67).
- **Before the final commit**: a full `./mvnw clean install` is not required
for this module, but the
beam module must build with Spotless on (`./mvnw -o -pl
plugins/engines/beam spotless:check`).
- **Manual verification that unit tests cannot cover**: run the BigQuery,
Kafka-Avro and Flink
changes once against a real cluster before claiming those sub-issues
fixed. If that is not
possible, say so on the issue rather than declaring victory.
## Risks, tradeoffs, and open questions
**Risks**
- *Scope.* 18 sub-issues is roughly 4–8 weeks of work. Waves 0, 2, 5 are the
cheap 20% and should
be finished first; wave 4 alone is half the effort and touches five new
transforms.
- *pom growth.* Waves 3 and 4 add ~7 new Beam artifacts plus several
`hop-transform-*` provided
deps. Each one enlarges the fat jar and the Flink/Spark classpath, which
is where Beam
dependency-hell failures show up. Add them **one transform per commit** so
a bad merge is
bisectable.
- *Tests without infrastructure.* Most of these transforms need a live Kafka
/ MQTT / Splunk /
Snowflake / Debezium / Elasticsearch endpoint. The unit tests can only
cover metadata,
registration and conversion logic; the IO wiring needs a Direct-runner
test at best and a real
cluster at worst. Be honest about which sub-issues are actually verified.
- *Fat-jar regressions.* T2.5a changes what the fat jar contains.
`FatJarBuilder` de-duplicates by
jar base name and skips classes already seen, so adding the Confluent jars
can shift which copy
of a shared class wins. Re-run `HopBeamGuiPluginNativeFatJarTest` and, if
a Flink sample is
available, one Flink job.
- *Dialog conventions.* `AGENTS.md` forbids hand-laying-out new
FormAttachment rows on the shell
for a dialog, but the existing Beam dialogs (`BeamBQInputDialog`,
`BeamOutputDialog`) are all
hand-laid-out. For **new** dialogs (waves 3–4) use `@GuiWidgetElement` +
`GuiCompositeWidgets`
with a `groupType`; for **extending** an existing hand-laid-out dialog
(T2.2, T2.3, T3.3) follow
that file's own local pattern so the dialog stays internally consistent.
Do not mix the two in
one dialog.
- *Spotless.* Three-attempt rule: if formatting fights you three times, run
`./mvnw spotless:apply` and move on rather than hand-tuning.
**Open questions — resolve before starting wave 4**
1. **#2613 scope.** The issue asks for "Transforms *and* Beam handlers for
Elastic". Is the Hop-side
Elastic transform part of this umbrella, or only the Beam handler? This
doubles T4.5's effort.
2. **#2378 / #2377 / #2375 licensing and support.** Snowflake and Splunk IO
pull vendor JDBC
drivers and have their own licensing. Confirm these belong in the Apache
distribution before
spending the effort.
3. **#2409 Debezium is unbounded-only.** A `DebeziumIO` transform is a
streaming source, so the
transform can only be an input and only works with the Flink/Spark
streaming runners. Confirm
that is acceptable, or drop the sub-issue.
4. **#2040 UI.** Should the partitioning UI be disabled on Beam pipelines
(it is currently ignored),
or should the dialogs explain that it is honoured per the new field hash?
Pick one and be
consistent.
5. **Backwards compatibility for #5064.** Mapping RECORD to a String changes
the output row type for
anyone who currently works around the crash by casting. Release-note it.
6. **#3131 / #3478** — confirm with the maintainer that closing a *question*
issue with a detailed
answer is acceptable, or whether they want a docs page instead. If the
latter, the content from
T5.2/T5.3 moves into
`docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/beam-flink-pipeline-engine.adoc`
and the issues get a one-line pointer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
