[FLINK-22470] The root cause of the exception encountered during compiling the job was not exposed to users in certain cases

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.11.4, 1.14.0, 1.13.1, 1.12.4
• Component/s: API / Python
• Labels: pull-request-available
• Assignee: Dian Fu
• Reporter: Dian Fu
• Created: 26/Apr/21 07:20
• Updated: 25/Jun/21 05:45
• Resolved: 27/Apr/21 02:26

Description

For the following job:

def test():
    from pyflink.table import DataTypes, BatchTableEnvironment, EnvironmentSettings

    env_settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()
    table_env = BatchTableEnvironment.create(environment_settings=env_settings)

    table_env \
        .get_config() \
        .get_configuration() \
        .set_string(
            "pipeline.jars",
            "file:///Users/dianfu/code/src/alibaba/ververica-connectors/flink-sql-avro-1.12.0.jar"
    )

    table = table_env.from_elements(
        [('111', '222')],
        schema=DataTypes.ROW([
            DataTypes.FIELD('text', DataTypes.STRING()),
            DataTypes.FIELD('text1', DataTypes.STRING())
        ])
    )
    sink_ddl = f"""
        create table Results(
            a STRING,
            b STRING
        ) with (
            'connector' = 'filesystem',
            'path' = '/Users/dianfu/tmp/',
            'format' = 'avro'
        )
    """

    table_env.execute_sql(sink_ddl)
    table.execute_insert("Results").wait()


if __name__ == "__main__":
    test()

It throws the following exception:

pyflink.util.exceptions.TableException: Failed to execute sql
	 at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:699)
	 at org.apache.flink.table.api.internal.TableImpl.executeInsert(TableImpl.java:572)
	 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	 at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	 at java.lang.reflect.Method.invoke(Method.java:498)
	 at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	 at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	 at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
	 at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	 at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
	 at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
	 at java.lang.Thread.run(Thread.java:748)

Process finished with exit code 1

The root cause isn't exposed and it's difficult for users to figure out what happens.

Issue Links

• links to: GitHub Pull Request #15765
• links to: GitHub Pull Request #15766