Abacn commented on code in PR #38892:
URL: https://github.com/apache/beam/pull/38892#discussion_r3537089265
##########
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:
There are different places we can put this test in. Is there a reason
prevent putting it in the base `FnApiRunnerTest` where most validate runner
tests live in?
##########
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:
I wonder whether we should move UnboundedCountingSource to the main
unbounded_source.py as it's used elsewhere. We can note it's internal in its
pydoc
--
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]
