Managed Apache Flink Does Not Work with Side Outputs

Below is an improved version of your message with added clarity and structure:

Hello,

I'm encountering an issue with my PyFlink processing pipeline that uses side output. The pipeline works correctly when run locally but fails on AWS Managed Apache Flink. I’ve created a minimal example to reproduce the error.

Minimal Code Example with Side Output

from pyflink.datastream import StreamExecutionEnvironment, KeyedProcessFunction, OutputTag
from pyflink.common import Types

side_output_tag = OutputTag("side-output", Types.STRING())

class MyProcessFunction(KeyedProcessFunction):
    def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
        yield value
        if value % 2 == 0:
            yield side_output_tag, f"Side output for even value: {value}"

def main():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)

    data = env.from_collection(collection=[1, 2, 3, 4, 5, 6], type_info=Types.INT())
    processed = data.key_by(lambda x: 1).process(MyProcessFunction(), output_type=Types.INT())
    processed.print("main")

    side_data = processed.get_side_output(side_output_tag)
    side_data.print("side")

    env.execute("Side Output Test")

if __name__ == "__main__":
    main()

Comparison: Code Without Side Output

This version, which does not use side output, runs perfectly locally and on AWS Managed Flink:

from pyflink.datastream import StreamExecutionEnvironment, KeyedProcessFunction
from pyflink.common import Types

class MyProcessFunction(KeyedProcessFunction):
    def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
        yield value

def main():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)

    data = env.from_collection(collection=[1, 2, 3, 4, 5, 6], type_info=Types.INT())
    processed = data.key_by(lambda x: 1).process(MyProcessFunction(), output_type=Types.INT())
    processed.print("main")

    env.execute("Side Output Test")

if __name__ == "__main__":
    main()

Build and Deployment Details

• Build Method: I’m using the pom.xml and assembly.xml in aws-samples/amazon-managed-service-for-apache-flink-examples GettingStarted repository to build the fat JAR; I simply replaced the main.py with the code above.
• Managed Flink Runtime: 1.20
• Java Version: openjdk 11.0.26
• Python Version: 3.11.11
• pip's apache-flink Package Version: 1.20.1

Exception Details

The error on Managed Flink can be viewed here: Exception Message.

Could anyone provide insight into why the side output functionality might be failing on AWS Managed Flink, or suggest any configuration or compatibility issues to investigate further?

Thank you!