dianfu commented on code in PR #20276:
URL: https://github.com/apache/flink/pull/20276#discussion_r928771352
##########
flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java:
##########
@@ -149,8 +151,8 @@ private static void
processSideOutput(List<Transformation<?>> transformations) {
final Transformation<?> upTransform =
Iterables.getOnlyElement(sideTransform.getInputs());
if (PythonConfigUtil.isPythonDataStreamOperator(upTransform)) {
- final AbstractDataStreamPythonFunctionOperator<?>
upOperator =
- (AbstractDataStreamPythonFunctionOperator<?>)
+ final AbstractExternalDataStreamPythonFunctionOperator<?>
upOperator =
Review Comment:
DataStreamPythonFunctionOperator?
##########
flink-python/pyflink/datastream/tests/test_data_stream.py:
##########
@@ -1675,6 +1678,101 @@ def flat_map_func2(data):
self.assert_equals_sorted(expected, results)
[email protected](sys.version_info < (3, 7), reason="requires python3.7")
+class EmbeddedDataStreamTests(PyFlinkStreamingTestCase):
+ def setUp(self):
+ super(EmbeddedDataStreamTests, self).setUp()
+ config =
get_j_env_configuration(self.env._j_stream_execution_environment)
+ config.setString("python.execution-mode", "thread")
+ config.setString("akka.ask.timeout", "20 s")
+ self.test_sink = DataStreamTestSinkFunction()
+
+ def tearDown(self) -> None:
+ self.test_sink.clear()
+
+ def assert_equals_sorted(self, expected, actual):
+ expected.sort()
+ actual.sort()
+ self.assertEqual(expected, actual)
+
+ def test_basic_operations(self):
Review Comment:
Could we refactor the test cases a bit and move test cases to be executed in
both process mode and thread mode in one class?
##########
flink-python/src/main/java/org/apache/flink/python/util/PythonConfigUtil.java:
##########
@@ -289,7 +291,9 @@ private static boolean isPythonDataStreamOperator(
StreamOperatorFactory<?> streamOperatorFactory) {
if (streamOperatorFactory instanceof SimpleOperatorFactory) {
return ((SimpleOperatorFactory<?>)
streamOperatorFactory).getOperator()
- instanceof AbstractDataStreamPythonFunctionOperator;
+ instanceof
AbstractExternalDataStreamPythonFunctionOperator
+ || ((SimpleOperatorFactory<?>)
streamOperatorFactory).getOperator()
Review Comment:
instance of DataStreamPythonFunctionOperator
--
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]
