[FLINK-30913] PyFlinkStreamUserDefinedFunctionTests.test_chaining_scalar_function failed due to missing response from the Java side

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.15.3
• Component/s: API / Python
• Labels: auto-deprioritized-critical test-stability
• Assignee: Unassigned
• Reporter: Matthias Pohl
• Created: 06/Feb/23 08:13
• Updated: 21/Aug/23 11:04
• Resolved: 21/Aug/23 11:04

Description

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=45731&view=logs&j=821b528f-1eed-5598-a3b4-7f748b13f261&t=6bb545dd-772d-5d8c-f258-f5085fba3295&l=29792

Feb 06 04:45:58 =================================== FAILURES ===================================
Feb 06 04:45:58 _____ PyFlinkStreamUserDefinedFunctionTests.test_chaining_scalar_function ______
Feb 06 04:45:58 
Feb 06 04:45:58 self = <pyflink.table.tests.test_udf.PyFlinkStreamUserDefinedFunctionTests testMethod=test_chaining_scalar_function>
Feb 06 04:45:58 
Feb 06 04:45:58     def test_chaining_scalar_function(self):
Feb 06 04:45:58         add_one = udf(lambda i: i + 1, result_type=DataTypes.BIGINT())
Feb 06 04:45:58         subtract_one = udf(SubtractOne(), result_type=DataTypes.BIGINT())
Feb 06 04:45:58     
Feb 06 04:45:58         table_sink = source_sink_utils.TestAppendSink(
Feb 06 04:45:58             ['a', 'b', 'c'],
Feb 06 04:45:58             [DataTypes.BIGINT(), DataTypes.BIGINT(), DataTypes.INT()])
Feb 06 04:45:58         self.t_env.register_table_sink("Results", table_sink)
Feb 06 04:45:58     
Feb 06 04:45:58         t = self.t_env.from_elements([(1, 2, 1), (2, 5, 2), (3, 1, 3)], ['a', 'b', 'c'])
Feb 06 04:45:58 >       t.select(add(add_one(t.a), subtract_one(t.b)), t.c, expr.lit(1)) \
Feb 06 04:45:58             .execute_insert("Results").wait()
Feb 06 04:45:58 
Feb 06 04:45:58 pyflink/table/tests/test_udf.py:89: 
Feb 06 04:45:58 _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
Feb 06 04:45:58 pyflink/table/table.py:1088: in execute_insert
Feb 06 04:45:58     return TableResult(self._j_table.executeInsert(table_path_or_descriptor, overwrite))
Feb 06 04:45:58 .tox/py38-cython/lib/python3.8/site-packages/py4j/java_gateway.py:1321: in __call__
Feb 06 04:45:58     return_value = get_return_value(
Feb 06 04:45:58 pyflink/util/exceptions.py:146: in deco
Feb 06 04:45:58     return f(*a, **kw)
Feb 06 04:45:58 _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
Feb 06 04:45:58 
Feb 06 04:45:58 answer = 'x'
Feb 06 04:45:58 gateway_client = <py4j.java_gateway.GatewayClient object at 0x7fef18871160>
Feb 06 04:45:58 target_id = 'o21847', name = 'executeInsert'
Feb 06 04:45:58 
Feb 06 04:45:58     def get_return_value(answer, gateway_client, target_id=None, name=None):
Feb 06 04:45:58         """Converts an answer received from the Java gateway into a Python object.
Feb 06 04:45:58     
Feb 06 04:45:58         For example, string representation of integers are converted to Python
Feb 06 04:45:58         integer, string representation of objects are converted to JavaObject
Feb 06 04:45:58         instances, etc.
Feb 06 04:45:58     
Feb 06 04:45:58         :param answer: the string returned by the Java gateway
Feb 06 04:45:58         :param gateway_client: the gateway client used to communicate with the Java
Feb 06 04:45:58             Gateway. Only necessary if the answer is a reference (e.g., object,
Feb 06 04:45:58             list, map)
Feb 06 04:45:58         :param target_id: the name of the object from which the answer comes from
Feb 06 04:45:58             (e.g., *object1* in `object1.hello()`). Optional.
Feb 06 04:45:58         :param name: the name of the member from which the answer comes from
Feb 06 04:45:58             (e.g., *hello* in `object1.hello()`). Optional.
Feb 06 04:45:58         """
Feb 06 04:45:58         if is_error(answer)[0]:
Feb 06 04:45:58             if len(answer) > 1:
Feb 06 04:45:58                 type = answer[1]
Feb 06 04:45:58                 value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
Feb 06 04:45:58                 if answer[1] == REFERENCE_TYPE:
Feb 06 04:45:58                     raise Py4JJavaError(
Feb 06 04:45:58                         "An error occurred while calling {0}{1}{2}.\n".
Feb 06 04:45:58                         format(target_id, ".", name), value)
Feb 06 04:45:58                 else:
Feb 06 04:45:58                     raise Py4JError(
Feb 06 04:45:58                         "An error occurred while calling {0}{1}{2}. Trace:\n{3}\n".
Feb 06 04:45:58                         format(target_id, ".", name, value))
Feb 06 04:45:58             else:
Feb 06 04:45:58 >               raise Py4JError(
Feb 06 04:45:58                     "An error occurred while calling {0}{1}{2}".
Feb 06 04:45:58                     format(target_id, ".", name))
Feb 06 04:45:58 E               py4j.protocol.Py4JError: An error occurred while calling o21847.executeInsert
Feb 06 04:45:58 
Feb 06 04:45:58 .tox/py38-cython/lib/python3.8/site-packages/py4j/protocol.py:334: Py4JError
Feb 06 04:45:58 ------------------------------ Captured log call -------------------------------
Feb 06 04:45:58 ERROR    root:java_gateway.py:1055 Exception while sending command.
Feb 06 04:45:58 Traceback (most recent call last):
Feb 06 04:45:58   File "/__w/1/s/flink-python/.tox/py38-cython/lib/python3.8/site-packages/py4j/java_gateway.py", line 1224, in send_command
Feb 06 04:45:58     raise Py4JNetworkError("Answer from Java side is empty")
Feb 06 04:45:58 py4j.protocol.Py4JNetworkError: Answer from Java side is empty
[...]

Issue Links

• duplicates: FLINK-30630 Python: BatchModeDataStreamTests.test_keyed_map failed