github-actions[bot] commented on code in PR #41285:
URL: https://github.com/apache/beam/pull/41285#discussion_r4560934089
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java:
##########
@@ -567,6 +568,10 @@ public BeamFnApi.InstructionResponse.Builder
processBundle(InstructionRequest re
finishFunction.run();
}
+ // Surface downstream exceptions that user code may have caught
while processing
+ // fused outputs. This must happen before outbound data is finalized.
+
bundleProcessor.getpCollectionConsumerRegistry().failBundleIfDownstreamFailed();
Review Comment:
**🟠High**: `failBundleIfDownstreamFailed()` is called *after* all
`@FinishBundle` functions have executed (lines 566-569). In `FAIL` mode, if a
downstream `DoFn` threw an exception during `@ProcessElement` that was
swallowed by an upstream `DoFn`, its `@FinishBundle` method will still be
invoked. This violates the `DoFn` lifecycle contract (finish bundle should not
be called on a failed transform instance), can cause downstream transforms to
flush corrupted/partial state to external sinks, or can throw a secondary
exception that masks `firstDownstreamException`.
Consider moving
`bundleProcessor.getpCollectionConsumerRegistry().failBundleIfDownstreamFailed();`
to immediately *before* the `finishFunctionRegistry` loop (or skipping finish
functions when a downstream failure has already been recorded in `FAIL` mode).
##########
sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java:
##########
@@ -935,6 +939,277 @@ public void testPTransformStartExceptionsArePropagated() {
assertThat(handler.bundleProcessorCache.getCachedBundleProcessors().get("1L"),
empty());
}
+ private static final AtomicInteger FUSED_DOWNSTREAM_INVOCATIONS = new
AtomicInteger();
+ private static final AtomicInteger FUSED_UPSTREAM_CAUGHT_EXCEPTIONS = new
AtomicInteger();
+
+ private static final class CatchingFusedDoFn extends DoFn<String, String> {
+ private static final TupleTag<String> MAIN_OUTPUT_TAG = new
TupleTag<>("mainOutput");
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ try {
+ context.output(context.element());
+ } catch (RuntimeException exception) {
+ FUSED_UPSTREAM_CAUGHT_EXCEPTIONS.incrementAndGet();
+ }
+ }
+ }
+
+ private static final class ThrowingFusedDoFn extends DoFn<String, String> {
+ private static final TupleTag<String> MAIN_OUTPUT_TAG = new
TupleTag<>("mainOutput");
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ FUSED_DOWNSTREAM_INVOCATIONS.incrementAndGet();
+ throw new IllegalStateException("fused downstream failure");
+ }
+ }
+
+ private static PTransform fusedDoFnTransform(
+ String inputPCollection,
+ String outputPCollection,
+ DoFnWithExecutionInformation doFnWithExecutionInformation) {
+ RunnerApi.FunctionSpec doFnSpec =
+ RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(ParDoTranslation.CUSTOM_JAVA_DO_FN_URN)
+ .setPayload(
+ ByteString.copyFrom(
+
SerializableUtils.serializeToByteArray(doFnWithExecutionInformation)))
+ .build();
+ ParDoPayload payload = ParDoPayload.newBuilder().setDoFn(doFnSpec).build();
+ return PTransform.newBuilder()
+ .setSpec(
+ FunctionSpec.newBuilder()
+ .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
+ .setPayload(payload.toByteString()))
+ .putInputs("input", inputPCollection)
+ .putOutputs("mainOutput", outputPCollection)
+ .build();
+ }
+
+ private ProcessBundleHandler setupHandlerForFusedExceptionTest(String
exceptionHandling)
+ throws Exception {
+ TupleTag<String> upstreamTag = CatchingFusedDoFn.MAIN_OUTPUT_TAG;
+ TupleTag<String> downstreamTag = ThrowingFusedDoFn.MAIN_OUTPUT_TAG;
+ DoFnWithExecutionInformation upstream =
+ DoFnWithExecutionInformation.of(
+ new CatchingFusedDoFn(),
+ upstreamTag,
+ Collections.emptyMap(),
+ DoFnSchemaInformation.create());
+ DoFnWithExecutionInformation downstream =
+ DoFnWithExecutionInformation.of(
+ new ThrowingFusedDoFn(),
+ downstreamTag,
Review Comment:
**🟢 Nit**: `FUSED_DOWNSTREAM_INVOCATIONS` and
`FUSED_UPSTREAM_CAUGHT_EXCEPTIONS` are static fields shared across tests. While
each test resets them at the start, resetting them in an `@Before` method or
scoping counters per-test instance avoids potential cross-test pollution if
tests are ever run concurrently or if new tests are added.
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/data/PCollectionConsumerRegistry.java:
##########
@@ -61,15 +63,29 @@
/**
* The {@code PCollectionConsumerRegistry} is used to maintain a collection of
consuming
* FnDataReceiver for each pCollectionId. Registering with this class allows
inserting an element
- * count counter for every pCollection. A combined MultiplexingConsumer
(Wrapped with an
- * ElementCountFnDataReceiver) is returned by calling getMultiplexingConsumer.
+ * count counter for every pCollection. A combined MultiplexingConsumer
(wrapped with metrics
+ * tracking) is returned by calling getMultiplexingConsumer. When configured,
this class also
+ * records downstream exceptions so that exceptions swallowed by fused user
code can be surfaced
+ * when the bundle finishes.
*/
@SuppressWarnings({
"rawtypes", // TODO(https://github.com/apache/beam/issues/20447)
"nullness" // TODO(https://github.com/apache/beam/issues/20497)
})
public class PCollectionConsumerRegistry {
+ /** Controls how exceptions from fused downstream consumers are handled at
bundle completion. */
+ public enum FusedDoFnExceptionHandling {
+ LEGACY,
+ WARN,
+ FAIL
+ }
+
+ private final FusedDoFnExceptionHandling fusedDoFnExceptionHandling;
+ private final AtomicReference<Exception> firstDownstreamException = new
AtomicReference<>();
Review Comment:
**🟡 Medium**: Downstream transforms may also throw `Error` (such as
`AssertionError` in tests or user validations), which can be caught by overly
broad `catch (Throwable t)` blocks in upstream user code. Because
`MetricTrackingFnDataReceiver` and `MultiplexingMetricTrackingFnDataReceiver`
only `catch (Exception e)`, any `Error` thrown by a downstream `DoFn` bypasses
`recordDownstreamException` (and `logAndRethrow`). Consider tracking
`Throwable` (or catching `Throwable` and rethrowing `Error` after recording) so
swallowed `AssertionError`s and other `Error`s are also surfaced at bundle
completion.
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/data/PCollectionConsumerRegistry.java:
##########
@@ -461,9 +554,13 @@ public void accept(WindowedValue<T> input) throws
Exception {
for (int size = consumerAndMetadatas.size(), i = 0; i < size; ++i) {
ConsumerAndMetadata consumerAndMetadata = consumerAndMetadatas.get(i);
ExecutionState state = consumerAndMetadata.getExecutionState();
+ // The original failure was already logged when the consumer first
threw. Avoid logging
+ // the same failure again for every later element rejected by
fail-fast handling.
+ failFastIfDownstreamFailed(consumerAndMetadata.getPTransformId());
Review Comment:
**🟠High**: When a `PCollection` has multiple consumers (e.g., `[Consumer1,
Consumer2]`), if `Consumer1` fails on element 1 and the upstream `DoFn` catches
the exception and emits element 2, `failFastIfDownstreamFailed` throws
immediately at `i = 0`. This aborts the loop before `Consumer2` ever receives
element 1 or element 2, even in a scenario where `WARN` vs `FAIL` behavior or
sibling consumer execution is expected before rethrowing to the caller.
Moreover, `this.elementCountCounter.inc(...)` on line 540 has already
incremented the PCollection counter before the fail-fast check in
`MetricTrackingFnDataReceiver` (where `failFastIfDownstreamFailed` runs before
`elementCountCounter.inc`), creating an inconsistency in metric reporting
between single and multiplexing consumers.
--
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]
