[Bug] Beam errors out when using PortableRunner (Flink Runner) – `Cannot run program "docker"`

[harrymyburgh]

What happened?

I am trying to deploy a Beam job (Python Beam) that runs on a PortableRunner (Flink Runner) in my Kubernetes cluster.
I have not experienced issues prior with Beam using the Flink Runner. However, today I tried to set up Beam to be a consumer from Apache Kafka using ReadFromKafka from apache_beam.io.kafka.

My Flink Cluster is managed by the Apache Flink Kubernetes Operator.

My Beam jobs are managed by a Beam Flink Job Manager, which posts Beam jobs to the Flink master. The Job Manager uses the image apache/beam_flink1.16_job_server:2.54.0.

My Flink Task Managers each contain a sidecar for a Beam worker pool, which is spun up using the image apache/beam_python3.11_sdk:2.54.0 and the arg --worker_pool.

When I start my beam job, I get the following error on the job manager logs:

Caused by: java.io.IOException: Cannot run program "docker": error=2, No such file or directory

These are my Beam pipeline options:

--job_name=beam_example_pipeline
--runner=PortableRunner
--job_endpoint=beam-flink-job-server:8099
--artifact_endpoint=beam-flink-job-server:8098
--environment_type=EXTERNAL
--environment_config=localhost:50000
--parallelism=1
--streaming

Some resources I've found suggest that the Kafka transform has its own environment type which is set to (and overrides any environment you set?) --environment_type=DOCKER, which is what causes the issues. However, I could be wrong, so please say so if I am.

All of this taking place on a Kubernetes cluster, where, to my knowledge, Docker in Docker is not recommended. I do not want to use a PROCESS environment_type, I require EXTERNAL. How can I resolve this issue? Is this a bug with Beam?

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• 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

[felipecaputo]

I'm currently facing the same problem myself, using a beam yaml pipeline, when it comes to run the SQL transformation it intents to run a docker with java in flink-main-container

WARN  org.apache.beam.runners.fnexecution.environment.DockerCommand [] - Unable to pull docker image apache/beam_java17_sdk:2.56.0, cause: Cannot run program "docker": error=2, No such file or directory
ERROR org.apache.flink.runtime.operators.BatchTask                 [] - Error in task code:  CHAIN MapPartition (MapPartition at [3]Sql/BeamCoGBKJoinRel_77/Join.Impl/CoGroup.ExpandCrossProduct/{extractKeylhs, CoGroupByKey}) -> FlatMap (FlatMap at ExtractOutput[0]) (1/1)
java.lang.Exception: The user defined 'open()' method caused an exception: java.io.IOException: Cannot run program "docker": error=2, No such file or directory
org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:508) ~[flink-dist-1.17.2.jar:1.17.2]
org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:357) ~[flink-dist-1.17.2.jar:1.17.2]
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) [flink-dist-1.17.2.jar:1.17.2]
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) [flink-dist-1.17.2.jar:1.17.2]
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) [flink-dist-1.17.2.jar:1.17.2]
org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) [flink-dist-1.17.2.jar:1.17.2]
java.lang.Thread.run(Unknown Source) [?:?]

The tasks were executed in the python-harness container runned smoothly, but when it come to the Sql transformation, it failed

[Raaaaaaaay86]

Facing same problem. Does anyone has solution? I'm trying to run a basic Kafka pipeline on my M1 MacBook.

This is the docker-compose which I'm running as an environment.

version: v3

services:
  kafka-1:
    container_name: my_kafka_cluster_node_1
    image: 'bitnami/kafka:3.5.1'
    deploy:
      resources:
        limits:
          memory: 2G
    networks:
      my_network:
        ipv4_address: 175.26.0.5
    ports:
      - 9097:9097
    environment:
      - KAFKA_KRAFT_CLUSTER_ID=CqhzvKYNSvmn-8NmZWgOHb
      - KAFKA_ENABLE_KRAFT=true
      - KAFKA_CFG_NODE_ID=1
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@175.26.0.5:9093
      - KAFKA_CFG_LISTENERS=INTERNAL://175.26.0.5:9092,CONTROLLER://175.26.0.5:9093,EXTERNAL://:9097
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://175.26.0.5:9092,EXTERNAL://localhost:9097
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL

  flink-jobmanager:
    image: flink:1.18
    networks:
      my_network:
        ipv4_address: 175.26.0.23
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        jobmanager.memory.process.size: 5gb

  flink-taskmanager:
    image: flink:1.18
    networks:
      my_network:
        ipv4_address: 175.26.0.24
    depends_on:
      - flink-jobmanager
    command: taskmanager
    scale: 1
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        taskmanager.numberOfTaskSlots: 20
        taskmanager.memory.process.size: 5gb
  
  beam-worker-pool:
    image: apache/beam_python3.11_sdk:latest
    networks:
      my_network:
        ipv4_address: 175.26.0.25
    ports:
      - "50000:50000"
    command: --worker_pool
    environment:
      - BEAM_WORKER_POOL_IN_DOCKER_VM=1

networks:
  my_network:
    driver: bridge
    ipam:
      driver: default
      config:
        - subnet: 175.26.0.0/24

and I run a beam_flink1.18_job_server seperatly by docker command.

docker run -d --network my_network -p 8099:8099 -p 8098:8098 -p 8097:8097 -v /tmp/f-conf:/flink-conf apache/beam_flink1.18_job_server:latest --flink-conf-dir=/flink-conf --flink-master=cre-center-flink-jobmanager:8081

and this is my Beam pipeline

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.kafka import ReadFromKafka

if __name__ == "__main__":
  options = PipelineOptions([
    "--runner=PortableRunner",
    "--job_endpoint=localhost:8099",
    "--environment_type=EXTERNAL",
    "--environment_config=175.26.0.25:50000"
  ])

  with beam.Pipeline(options=options) as pipeline:
    (
    pipeline
    | "Read Data" >> ReadFromKafka(
      consumer_config={
        "bootstrap.servers": "175.26.0.5:9092",
        "auto.offset.reset": "earliest",
      },
      topics=["member_entered_store"])
    )

finally, I'll execute the pipeline by running command on MacBook

python beam_playground/main.py

and I receive same error in beam_flink1.18_job_server container

Cannot run program "docker": error=2, No such file or directory

[liferoad]

I think the flink cluster should have the docker installed since it needs to download the Python SDK worker image.

[mpiraino]

These configurations:

--environment_type=EXTERNAL
--environment_config=175.26.0.25:50000

specify the location where beam will run your python code. ReadFromKafka is made available in the python sdk through a java expansion service and you need to override the default java expansion service environment type by doing something like:

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.kafka import ReadFromKafka, default_io_expansion_service

if __name__ == "__main__":
  options = PipelineOptions([
    "--runner=PortableRunner",
    "--job_endpoint=localhost:8099",
    "--environment_type=EXTERNAL",
    "--environment_config=175.26.0.25:50000"
  ])

  with beam.Pipeline(options=options) as pipeline:
    (
    pipeline
    | "Read Data" >> ReadFromKafka(
      consumer_config={
        "bootstrap.servers": "175.26.0.5:9092",
        "auto.offset.reset": "earliest",
      },
      topics=["member_entered_store"]),
      expansion_service=default_io_expansion_service(
           append_args=[
                '--defaultEnvironmentType=PROCESS',
                '--defaultEnvironmentConfig={"command":"/opt/apache/beam/boot"}',
            ]
       ),
    )

Note that using the using the process type requires you to have the boot executable in the flink task manager. I think that theoretically it is possible to specify EXTERNAL provided you have a pool of java sdk harness workers (I have never actually tried to set up one of those so not sure how that works, but would be very interested to see an example if someone makes it work).

[cliffsun91]

I've come across this problem but I'm attempting to use Beam YAML and it specifically only happens when I try to use a custom provider (I copied the example from https://github.com/apache/beam-starter-python-provider and using the Stringify transform which isn't doing anything special). I'm also running the SDK harness as a sidecar container along side the Flink TaskManager container (in a pod in local k8s) and specify the --environment_type=EXTERNAL. I'm not too sure why it only happens when I attempt to use a custom provider transform - is this a known issue?

[marlonklobukoski-aquarela]

Hi @cliffsun91 i'm facing the same issue. Did you find any solution?

[yuwtennis]

I just suffered same problem and able to reproduce in 2.69.0 using PortableRunner w/ Apache Flink backend.
I would also would like to know if there is an workaround.

It looks like the behavior in the ExecutableStage is different with bounded and unbounded data source.

For case with bounded data source, the --environment_type is respected . Below is the snippet from the Flink taskmanager .

sample run

2025-11-27 12:13:22,207 DEBUG org.apache.beam.runners.fnexecution.environment.ExternalEnvironmentFactory [] - Requesting worker ID 1-1

However, for unbounded data source, following the stacktrace it is actually hardcoded using DOCKER environment.

sample run

2025-11-27 12:25:37,368 DEBUG org.apache.beam.runners.fnexecution.environment.DockerEnvironmentFactory [] - Creating Docker Container with ID 1-1

Below are the logs.

bounded.txt
unbounded.txt