[Docs] Add a contributor guide for running Python on a local Flink cluster

contributor-docs/README.md

@@ -22,6 +22,7 @@ This directory contains documentation for contributors to the Apache Beam projec
 - [Committer Guide](committer-guide.md): Guidelines for Beam committers regarding code review, pull request objectives, merging processes, and post-merge tasks.
 - [Committer Onboarding](committer-onboarding.md): A checklist for new Beam committers to set up their accounts and permissions.
 - [Java Dependency Upgrades](java-dependency-upgrades.md): Instructions for upgrading Java dependencies in Beam, including running linkage checkers and verification tests.
+- [Local Flink Python Validation](local-flink-python.md): Instructions for running Python pipelines on a local Flink standalone cluster.
 - [Python Tips](python-tips.md): Tips and instructions for developing the Python SDK, including environment setup, running tests, and handling dependencies.
 - [RC Testing Guide](rc-testing-guide.md): A guide for testing Beam Release Candidates (RCs) against downstream projects for Python, Java, and Go SDKs.
 - [Release Guide](release-guide.md): A comprehensive guide for the Release Manager on how to perform a Beam release, from preparation to promotion.

contributor-docs/local-flink-python.md

@@ -0,0 +1,204 @@
+<!--
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+# Running Python pipelines on a local Flink cluster
+
+This guide describes a contributor workflow for validating Python Beam pipelines
+against a real local Flink standalone cluster. It is useful when embedded Flink
+is not enough, for example when validating streaming source behavior, checkpoint
+boundaries, or runner-visible job state in the Flink dashboard.
+
+The commands assume a Unix shell (Linux, macOS, or WSL2 on Windows) with `curl`,
+`tar`, and `java` on the `PATH`.
+
+* [What this setup validates](#what-this-setup-validates)
+* [Prerequisites](#prerequisites)
+* [Start a local Flink cluster](#start-a-local-flink-cluster)
+* [Run a Beam Python pipeline](#run-a-beam-python-pipeline)
+* [Troubleshooting](#troubleshooting)
+* [Stop the cluster](#stop-the-cluster)
+
+## What this setup validates
+
+This setup runs three separate processes:
+
+1. A Flink standalone cluster, consisting of a JobManager and a TaskManager.
+1. A Beam Flink Job Server, started by the Python `FlinkRunner`.
+1. A Python SDK harness, using `--environment_type=LOOPBACK` for local
+   development.
+
+The Flink dashboard at `http://localhost:8081` shows the submitted Beam jobs.
+This is different from embedded Flink mode, where the cluster is started only
+for the lifetime of one job and is not useful for manual dashboard inspection.
+
+## Prerequisites
+
+Install or prepare the following:
+
+* Docker Desktop (optional), only for the alternative method of obtaining the
+  Flink distribution.
+* A Unix shell: Linux, macOS, or WSL2 on Windows.
+* Java 11 on the `PATH`.
+* A Python environment with the Beam SDK dependencies installed.
+* A Beam source checkout for the Python code under test.
+* A Flink 1.20 Job Server jar built from the same Beam checkout when validating
+  unreleased Beam changes.
+
+For a source-built Job Server jar, run this command from the Beam checkout:
+
+```sh
+./gradlew :runners:flink:1.20:job-server:shadowJar
+```
+
+The jar is written under:
+
+```text
+runners/flink/1.20/job-server/build/libs/
+```
+
+## Start a local Flink cluster
+
+Use a Flink distribution whose minor version matches a Flink version supported
+by your Beam version. See the [Flink Version Compatibility](https://beam.apache.org/documentation/runners/flink/#flink-version-compatibility)
+table in the Flink Runner documentation, and confirm the exact patch version on
+the [Flink downloads page](https://flink.apache.org/downloads.html). This guide
+uses Flink 1.20.
+
+Download and unpack the binary distribution:
+
+```sh
+FLINK_VERSION=1.20.1
+curl -fLO "https://archive.apache.org/dist/flink/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-scala_2.12.tgz"
+tar -xzf "flink-${FLINK_VERSION}-bin-scala_2.12.tgz" -C "$HOME"
+export FLINK_HOME="$HOME/flink-${FLINK_VERSION}"
+```
+
+Ensure these settings exist in `$FLINK_HOME/conf/config.yaml`:
+
+```yaml
+jobmanager.rpc.address: localhost
+rest.address: localhost
+taskmanager.numberOfTaskSlots: 2
+```
+
+Start the cluster. The JobManager and TaskManager run as background daemons:
+
+```sh
+"$FLINK_HOME/bin/start-cluster.sh"
+```
+
+Verify that the JobManager and TaskManager are available:
+
+```sh
+curl -fsS http://localhost:8081/overview
+```
+
+Expected output includes one TaskManager and two slots:
+
+```json
+{"taskmanagers":1,"slots-total":2,"slots-available":2,"jobs-running":0}
+```
+
+You can also open the Flink dashboard in a browser:
+
+```text
+http://localhost:8081
+```
+
+### Alternative: extract Flink from the Docker image
+
+If a direct download is not available, copy the distribution out of the Flink
+Docker image with `docker cp`:
+
+```sh
+docker create --name flink-dist flink:1.20
+docker cp flink-dist:/opt/flink "$HOME/flink-1.20"
+docker rm flink-dist
+export FLINK_HOME="$HOME/flink-1.20"
+```
+
+A distribution copied out of a Docker image can contain the container hostname in
+`conf/config.yaml`; see [Troubleshooting](#troubleshooting).
+
+## Run a Beam Python pipeline
+
+For local Python development, use `FlinkRunner`, point it at the standalone
+cluster, and use `LOOPBACK` so the Python SDK harness runs in the local process.
+
+Use a source checkout on `PYTHONPATH` when validating unreleased Python changes.
+Set paths for your environment:
+
+```sh
+export BEAM_CHECKOUT="$HOME/beam"
+export PYTHON="$HOME/beamenv/bin/python"
+export FLINK_JOB_SERVER_JAR="$(find "$BEAM_CHECKOUT/runners/flink/1.20/job-server/build/libs" \
+  -name 'beam-runners-flink-1.20-job-server-*.jar' | head -n 1)"
+```
+
+Run a small pipeline:
+
+```sh
+printf 'to be or not to be\nbeam runs on flink\n' > /tmp/beam-flink-input.txt
+
+PYTHONPATH="$BEAM_CHECKOUT/sdks/python" "$PYTHON" -m apache_beam.examples.wordcount \
+  --runner=FlinkRunner \
+  --flink_master=localhost:8081 \
+  --flink_version=1.20 \
+  --flink_job_server_jar="$FLINK_JOB_SERVER_JAR" \
+  --environment_type=LOOPBACK \
+  --input=/tmp/beam-flink-input.txt \
+  --output=/tmp/beam-flink-counts
+```
+
+For released Beam, omit `--flink_job_server_jar` and the `PYTHONPATH` prefix; the
+`FlinkRunner` downloads a Job Server matching `--flink_version` automatically. The
+source checkout and built jar are only needed to test unreleased changes.
+
+Check the dashboard or REST API after the run:
+
+```sh
+curl -fsS http://localhost:8081/jobs/overview
+```
+
+The job should be `FINISHED`.
+
+## Troubleshooting
+
+If the TaskManager does not register, check `$FLINK_HOME/conf/config.yaml`.
+When a distribution is copied out of a Docker image, the file might contain the
+container hostname. Replace it with:
+
+```yaml
+jobmanager.rpc.address: localhost
+```
+
+If a Python job fails on native Windows with an invalid path containing `:`,
+run the Python driver and Job Server from WSL2. Some staged artifact names used
+by the portable runner are valid on Linux but invalid as native Windows file
+names.
+
+On WSL2, keep at least one shell open in the distribution while the cluster runs.
+Closing the last shell can stop the distribution and its background daemons.
+
+If the job starts but the Python transforms do not execute, check the
+environment type. `LOOPBACK` is intended for local development. For a remote
+or multi-machine Flink cluster, use a containerized environment instead.
+
+## Stop the cluster
+
+Stop the local cluster when you finish collecting results:
+
+```sh
+"$FLINK_HOME/bin/stop-cluster.sh"
+```