uros-b commented on code in PR #57653:
URL: https://github.com/apache/spark/pull/57653#discussion_r3683405978
##########
python/pyspark/sql/tests/streaming/test_streaming_foreach_batch.py:
##########
@@ -119,6 +119,24 @@ def func(df: DataFrame, batch_id: int):
)
self.assertEqual(sorted(df.collect()), sorted(actual.collect()))
+ def test_streaming_foreach_batch_operation_tags(self):
Review Comment:
The new test is added to StreamingTestsForeachBatchMixin, which is inherited
by both the Connect parity suite (StreamingForeachBatchParityTests) and the
classic suite (StreamingTestsForeachBatch). In classic mode the foreachBatch
callback runs on the Py4J callback server thread, a persistent pre-created
thread. Classic SparkSession.addTag() sets tags in the JVM via
InheritableThreadLocal; that Py4J thread was created before addTag() is called
in the test body, so the tag will not be visible in the callback. The assertion
assert tag in batch_df.sparkSession.getTags() will fail for the classic suite.
The test exercises Connect-specific behaviour (the Python subprocess
tag-injection mechanism) and belongs in
python/pyspark/sql/tests/connect/streaming/test_parity_foreach_batch.py, not in
the shared mixin.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
