Eliaaazzz commented on code in PR #38892:
URL: https://github.com/apache/beam/pull/38892#discussion_r3543320650
##########
sdks/python/apache_beam/runners/portability/portable_runner_test.py:
##########
@@ -209,6 +211,30 @@ def process(self, kv,
index=beam.DoFn.StateParam(index_state_spec)):
| beam.ParDo(AddIndex()),
equal_to(expected))
+ def test_unbounded_source_read(self):
+ """Reads a self-terminating UnboundedSource through the job service.
+
+ Complements the DirectRunner tests in apache_beam.io.unbounded_source_test
+ by exercising the runner-API round trip and the EOF ``MAX_TIMESTAMP``
+ watermark propagation that lets the downstream window fire.
+ """
+ if type(self).__name__ == 'SparkRunnerTest':
+ # Portable Spark does not execute SDFs, so SparkRunnerTest skips every
+ # SDF test (see its test_sdf* overrides).
Review Comment:
No reason not to, that is a better home for it. I moved
`test_unbounded_source_read` into the base `FnApiRunnerTest` so the in-process
`FnApiRunner` variants and the portable suites all inherit it. The multi-worker
and bundle-repeat variants skip it as single worker only, matching the sibling
SDF checkpointing and watermark-tracking tests, and `SparkRunnerTest` still
skips it for #19468.
##########
sdks/python/apache_beam/runners/portability/portable_runner_test.py:
##########
@@ -30,6 +30,7 @@
import grpc
import apache_beam as beam
+from apache_beam.io.unbounded_source_test import UnboundedCountingSource
Review Comment:
Good idea, done. I moved `UnboundedCountingSource` together with its reader
and checkpoint mark into `unbounded_source.py`, marked it internal in the
pydoc, and kept it out of `__all__`. `unbounded_source_test.py` and
`iobase_test.py` now import it from there.
--
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]
