junaiddshaukat opened a new pull request, #39451:
URL: https://github.com/apache/beam/pull/39451
Part of #18479.
Enables Beam's `ParDoTest` suites in the Kafka Streams runner's
`validatesRunner` task, which takes the suite from 26 to 44 tests.
### What runs now
| Suite | Tests |
| --- | --- |
| `ParDoTest$BasicTests` | 6 |
| `ParDoTest$MultipleInputsAndOutputTests` | 5 |
| `ParDoTest$TimestampTests` | 4 |
| `ParDoTest$LifecycleTests` | 3 |
`MultipleInputsAndOutputTests` passes on the multi-output executable stages
that landed in #39410. The remaining `ParDoTest` nested classes (state, timers,
timer families, bundle finalization, and the side-input half of
`MultipleInputsAndOutputTests`) fall out on category excludes the task already
declares, so no new excludes were needed.
### Surfacing a failed run as `PipelineExecutionException`
The three `LifecycleTests` error tests assert that a DoFn throwing surfaces
as a `RuntimeException` carrying the user's message. What came out of
`TestKafkaStreamsRunner.run` instead was the raw Kafka Streams wrapper:
```
java.lang.AssertionError:
Expected: (an instance of java.lang.RuntimeException and exception with
message a string containing "test error in process")
but: ... message was "stream-thread [Time-limited test] task [0_0]
Exception caught while punctuating processor 'Create-Values-Read-CreateSource-'"
```
The user's exception is at the bottom of that chain, under several layers of
bundle- and Fn-API-level wrappers. `run` now restates a failed run as
`Pipeline.PipelineExecutionException` carrying the root cause, so the user's
message ends up on the thrown exception — `RuntimeException(Throwable)` derives
its message from the cause. `SparkPipelineResult` and `DirectRunner` restate
failures the same way. The existing unwrap that rethrows an `AssertionError` (a
failed `PAssert`) still runs first.
That fixes `testParDoWithErrorInProcessElement` and
`testParDoWithErrorInFinishBatch`.
### One sickbay
`testParDoWithErrorInStartBatch` is sickbayed. A DoFn whose `@StartBundle`
throws never gets to report its error: `SdkHarnessClient.newBundle` sends the
`ProcessBundleRequest` and then blocks in
`GrpcDataService.createOutboundAggregator` waiting for the SDK harness to open
its data stream, which a bundle that failed during setup never does. The run
blocks for the data service's three-minute timeout instead of surfacing the
user's exception.
This is shared `java-fn-execution` behaviour rather than anything specific
to this runner — the Flink runner sickbays all of `LifecycleTests`, and the
Prism runner sickbays each of its three error tests. The `@ProcessElement` and
`@FinishBundle` variants do pass here, because by then the data stream is
established.
### Testing
```
./gradlew :runners:kafka-streams:validatesRunner # 44 tests, 0 failures
./gradlew :runners:kafka-streams:build # 61 unit tests, spotless
+ checker clean
```
--
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]
