What happened?

Please see https://gist.github.com/nybbles/6e1f2ab31866b251ff754e22b71f8405 for code to replicate this problem.

Problem description

I am finding that using unbounded SDFs with Flink results in checkpoint sizes that grow without bound, which eventually results in the job failing and all further job submissions to fail, even for beam.transforms.periodicsequence.PeriodicImpulse in a very simple pipeline, given below.

My pipeline consists of an SDF that reads from an unbounded source, which means that when there are no new messages, the SDF must poll the unbounded source, with some timeout. I observed that when my SDF would do this polling behavior (using tracker.defer_remainder as described in https://beam.apache.org/documentation/programming-guide/#user-initiated-checkpoint, the checkpoint size would grow.

This happens even if the unbounded source was empty, and hence my SDF simply executed a loop of polling the unbounded source and then calling tracker.defer_remainder and returning from the DoFn to relinquish control and wait to poll again.

I was concerned that I had implemented my SDF or my pipeline incorrectly, so I found beam.transforms.periodicsequence.PeriodicImpulse and tested it in a very simple pipeline, which is as follows (note that apply_windowing’s value does not change the problematic behavior):

    with beam.Pipeline(options=runner_options) as pipeline:
        pcoll = pipeline | PeriodicImpulse(
            fire_interval=5, apply_windowing=True
        )
        pcoll | beam.Map(print)

This pipeline also results in growing checkpoint size.

The Flink cluster configuration and full source for the program to replicate the problem and the Docker compose to get the Flink cluster up and running are given below in the reproduction steps and in https://gist.github.com/nybbles/6e1f2ab31866b251ff754e22b71f8405.

In case it is helpful, I’ll list the FLINK_PROPERTIES and PipelineOptions below.

      FLINK_PROPERTIES: &flink_properties |-
        historyserver.archive.clean-expired-jobs: true
        state.backend: hashmap
        state.checkpoints.dir: file:///tmp/beam_state/
        state.checkpoints.num-retained: 10
        jobmanager.rpc.address: host.docker.internal
        rest.address: host.docker.internal
        taskmanager.numberOfTaskSlots: 4
        taskmanager.memory.process.size: 2048m

    runner_options = PortableOptions(
        artifact_endpoint=f"{JOB_SERVER_HOSTNAME}:8098",
        environment_cache_millis=0,
        environment_config="apache/beam_python3.11_sdk:2.48.0",
        environment_options=None,
        environment_type="DOCKER",
        job_endpoint=f"{JOB_SERVER_HOSTNAME}:8099",
        job_server_timeout=60,
        runner = "PortableRunner",
        output_executable_path=None,
        sdk_worker_parallelism=0,
        state_backend = "filesystem",
        state_backend_storage_path = "file:///tmp/beam_state/",
        streaming = True,
        checkpointing_interval = STREAMING_CHECKPOINTING_INTERVAL,
        parallelism = 1,
        auto_watermark_interval = 500,
    )

See this email thread for more context: https://lists.apache.org/thread/7yjr1f24rdzwzofdty1h12w9m28o62sm.

Note on priority

I followed the linked guide for setting issue priorities and set this one to priority 1 because it seems like unbounded SDFs is an important component, running on Flink is an important usecase, and having arbitrary checkpoint size growth makes unbounded SDFs on Flink non-functional. My apologies in advance if this is the wrong priority level.

Reproduction steps

1. Run a Flink cluster (i.e. the Gist above provides my Docker Compose-based Flink cluster. It uses a taskmanager image that can run Docker containers, needed for using environment_type="DOCKER").
2. Run the attached driver program in the Gist above to create the job and submit it to the Flink cluster.
3. Observe in the logs that the checkpoints grow in size, specifically the step with the SDF, (and in proportion to the number of calls to tracker.defer_remainder), despite the SDF not actually explicitly accumulating any state.
4. The checkpoint size grows until the Java heap space is exhausted and the job is killed. Now when the job is resubmitted, it will always fail, because the job manager attempts to restore the job from the large checkpoint, resulting in Java heap space being exhausted again.

Issue Priority

Priority: 1 (data loss / total loss of function)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner