github-actions[bot] commented on code in PR #40909:
URL: https://github.com/apache/beam/pull/40909#discussion_r4427195663
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/LocalResolveDoFnTest.java:
##########
@@ -247,11 +251,42 @@ private List<ValueInSingleWindow<Row>> process(
List<SerializableChangelogTask> tasks,
Instant timestamp)
throws Exception {
- try (DoFnTester<KV<ChangelogDescriptor, List<SerializableChangelogTask>>,
Row> tester =
- DoFnTester.of(new LocalResolveDoFn(scanConfig))) {
- tester.processTimestampedElement(TimestampedValue.of(KV.of(descriptor,
tasks), timestamp));
- return tester.getMutableOutput(tester.getMainOutputTag());
- }
+ LocalResolveDoFn fn = new LocalResolveDoFn(scanConfig);
+ fn.setup();
+ List<ValueInSingleWindow<Row>> outputs = new ArrayList<>();
+ OutputReceiver<Row> receiver =
+ new OutputReceiver<Row>() {
+ @Override
+ public void output(Row output) {
+ outputWithTimestamp(output, timestamp);
+ }
+
+ @Override
+ public void outputWithTimestamp(Row output, Instant ts) {
+ outputs.add(
+ ValueInSingleWindow.of(output, ts, GlobalWindow.INSTANCE,
PaneInfo.NO_FIRING));
+ }
+
+ @Override
+ public OutputBuilder<Row> builder(Row output) {
+ return WindowedValues.<Row>builder()
+ .setValue(output)
+ .setWindow(GlobalWindow.INSTANCE)
+ .setPaneInfo(PaneInfo.NO_FIRING)
+ .setTimestamp(timestamp)
+ .setReceiver(
+ wv ->
+ outputs.add(
+ ValueInSingleWindow.of(
Review Comment:
**🔴 Critical**: `LocalResolveDoFn` emits CDC records using
`out.builder(row).setValueKind(kind).output()`, and the tests assert on
`value.getValueKind()` (via `kindAndProjectedRow`). However, the 4-argument
`ValueInSingleWindow.of(value, timestamp, window, paneInfo)` overload hardcodes
`ValueKind.INSERT`, discarding `wv.getValueKind()`. As a result, all emitted
records will have `ValueKind.INSERT` instead of `UPDATE_BEFORE`,
`UPDATE_AFTER`, or `DELETE`, causing the tests in this class to fail.
Use the 9-argument `ValueInSingleWindow.of(...)` overload and pass
`wv.getValueKind()` (and `wv.causedByDrain()`). You can also remove the
redundant `output` and `outputWithTimestamp` overrides since `OutputReceiver`
already provides default methods that delegate to `builder(...)`.
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ResolveChangesTest.java:
##########
@@ -175,20 +179,51 @@ private List<ValueInSingleWindow<Row>> process(
throws Exception {
CoGbkResult result =
CoGbkResult.of(ResolveChanges.DELETES,
deletes).and(ResolveChanges.INSERTS, inserts);
- try (DoFnTester<KV<CdcRowDescriptor, CoGbkResult>, Row> tester =
- DoFnTester.of(new ResolveChanges(scanConfig(icebergSchema)))) {
- tester.processTimestampedElement(
- TimestampedValue.of(
- KV.of(
- CdcRowDescriptor.builder()
- .setCommitSnapshotId(123)
- .setSnapshotSequenceNumber(456)
- .setPrimaryKey(pk)
- .build(),
- result),
- timestamp));
- return tester.getMutableOutput(tester.getMainOutputTag());
- }
+ ResolveChanges fn = new ResolveChanges(scanConfig(icebergSchema));
+ fn.setup();
+ List<ValueInSingleWindow<Row>> outputs = new ArrayList<>();
+ OutputReceiver<Row> receiver =
+ new OutputReceiver<Row>() {
+ @Override
+ public void output(Row output) {
+ outputWithTimestamp(output, timestamp);
+ }
+
+ @Override
+ public void outputWithTimestamp(Row output, Instant ts) {
+ outputs.add(
+ ValueInSingleWindow.of(output, ts, GlobalWindow.INSTANCE,
PaneInfo.NO_FIRING));
+ }
+
+ @Override
+ public OutputBuilder<Row> builder(Row output) {
+ return WindowedValues.<Row>builder()
+ .setValue(output)
+ .setWindow(GlobalWindow.INSTANCE)
+ .setPaneInfo(PaneInfo.NO_FIRING)
+ .setTimestamp(timestamp)
+ .setReceiver(
+ wv ->
+ outputs.add(
+ ValueInSingleWindow.of(
Review Comment:
**🔴 Critical**: Similar to `LocalResolveDoFnTest`, `ResolveChanges` sets the
CDC kind via `.setValueKind(kind)` on the `OutputBuilder`, and `kindAndData`
inspects `value.getValueKind()`. Calling the 4-argument
`ValueInSingleWindow.of(...)` overload drops `wv.getValueKind()` and defaults
to `ValueKind.INSERT`, which breaks
`updatePairAndExtraRowsPreserveKindsAndTimestamp`.
Pass `wv.getValueKind()` using the 9-argument
`ValueInSingleWindow.of(wv.getValue(), wv.getTimestamp(),
wv.getWindows().iterator().next(), wv.getPaneInfo(), wv.getRecordId(),
wv.getRecordOffset(), wv.causedByDrain(), wv.getOpenTelemetryContext(),
wv.getValueKind())` overload (and optionally drop the redundant
`output`/`outputWithTimestamp` overrides).
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/BundleLifterTest.java:
##########
@@ -36,64 +44,89 @@ public class BundleLifterTest {
private static final TupleTag<String> STRING_SMALL = new TupleTag<String>()
{};
private static final TupleTag<String> STRING_LARGE = new TupleTag<String>()
{};
+ private static class SimpleMultiOutputReceiver implements
MultiOutputReceiver {
+ private final Map<TupleTag<?>, TestOutputReceiver<?>> receivers = new
HashMap<>();
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public @UnknownKeyFor @NonNull @Initialized <T> OutputReceiver<T> get(
+ @UnknownKeyFor @NonNull @Initialized TupleTag<T> tag) {
+ return (OutputReceiver<T>) receivers.computeIfAbsent(tag, k -> new
TestOutputReceiver<T>());
+ }
+
+ @Override
+ public @UnknownKeyFor @NonNull @Initialized <T>
+ OutputReceiver<@UnknownKeyFor @NonNull @Initialized Row>
getRowReceiver(
+ @UnknownKeyFor @NonNull @Initialized TupleTag<T> tag) {
+ throw new UnsupportedOperationException("getRowReceiver not supported in
test");
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T> java.util.List<T> getOutput(TupleTag<T> tag) {
Review Comment:
**🟢 Nit**: Prefer importing `java.util.Collections` and `java.util.List` at
the top of the file rather than using fully-qualified names here and on line 67.
--
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]
