flink-local-dev

# Flink local development Validate Flink SQL logic where the code lives: one JVM, one MiniCluster, no Kubernetes, no Docker, no shared cluster. The job script that passes here is the same file that ships to production; everything local lives in a separate overlay. Copy the templates below into the project, replace every `__PLACEHOLDER__`, and delete the variants you do not use. They are starting points, not defaults. ## 1. Establish the target Collect or derive these before changing files: - the SQL script under validation, and which statements are the job (a single `INSERT INTO` or an `EXECUTE STATEMENT SET`) - the production Flink version this job will run on (`spec.flinkVersion` such as `v1_20` -> Flink `1.20.x`) - every source table and, for each, whether the local run uses a development endpoint or a generated stand-in - every sink table — each one needs a local shadow - the connector jars the script needs, and their versions - expected output: row count, sample rows, changelog kinds, or an invariant the result must satisfy Use this layout, next to the Operator manifests the `flink-k8s-operator` skill generates: ``` deploy/flink/<name>/ ├── sql/job.sql # the production artifact — identical locally └── local/ ├── local-session.sql # local session settings ├── local-sources.sql # bounded/dev source shadows (optional) ├── local-sinks.sql # sink shadows (required when the job writes) └── run-local-sql.sh # preflight + guards + SQL Client invocation ``` Inspect existing files first and never overwrite them silently. Ask which statements form the job when the script mixes DDL, DML, and ad-hoc queries. ## 2. Preflight the local runtime (fail closed) A local run needs a Flink distribution on this machine — `apache-flink` in a virtualenv is not one: ```bash echo "${FLINK_HOME:?set FLINK_HOME to a local Flink distribution}" "$FLINK_HOME/bin/flink" --version # -> Version: 1.20.1, Commit ID: ... java -version ``` Two invariants: - The distribution's minor version matches the production `flinkVersion`. A different minor version changes planner behaviour and connector compatibility, which defeats the point of validating locally. - The JDK is one that release supports. Flink 1.x releases target Java 8/11/17 and Flink 2.x dropped Java 8; check the release's own compatibility note rather than assuming. An unsupported JDK fails with `InaccessibleObjectException` or a module-access error, not a version message — point `JAVA_HOME` at a supported JDK instead of adding `--add-opens` flags. Then prove the in-process path works, before any real endpoint is involved: ```bash "$FLINK_HOME/bin/sql-client.sh" -f local/minicluster-smoke.sql ``` ### minicluster-smoke.sql ```sql -- Proves the in-process MiniCluster path works. Expect five +I[...] rows on -- stdout and exit code 0. Reads nothing, writes nothing, needs no connector jar. SET 'execution.target' = 'local'; SET 'table.dml-sync' = 'true'; SET 'parallelism.default' = '1'; SET 'sql-client.execution.result-mode' = 'tableau'; CREATE TEMPORARY TABLE local_smoke_source ( id BIGINT, label STRING ) WITH ( -- A sequence field bounds the table: it ends when the sequence completes. 'connector' = 'datagen', 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' = '5', 'fields.label.length' = '4' ); CREATE TEMPORARY TABLE local_smoke_sink ( id BIGINT, label STRING ) WITH ( 'connector' = 'print', 'print-identifier' = 'local_smoke_sink' ); INSERT INTO local_smoke_sink SELECT id, label FROM local_smoke_source; ``` `SET 'execution.target' = 'local'` selects Flink's `LocalExecutor`, which builds the JobGraph and submits it to a MiniCluster inside the client JVM: no JobManager or TaskManager daemon, nothing surviving the process, no `flink-conf.yaml` cluster address. The SQL Client documentation only describes the `start-cluster.sh` + remote-target workflow, so treat the local target as verified-by-probe — run the smoke script once per distribution before trusting it. Stop with an actionable message when `FLINK_HOME` is unset, the version does not match production, the JDK is unsupported, a connector factory is missing, or the smoke run fails. If a release rejects the local target, **do not** let the run fall through to the default target — with a real development source configured, that submits the job to whatever cluster `flink-conf.yaml` points at. Fix the version, or ask the user before starting a local standalone cluster (`"$FLINK_HOME/bin/start-cluster.sh"`, remote target, UI on `localhost:8081`) and stop it afterwards. ## 3. Bound the blast radius A local MiniCluster is isolated in compute only — its connectors reach exactly as far as the credentials you give them. Non-negotiable: 1. Sources may be real development endpoints; sinks may not. 2. Read-only credentials, and an identity of its own — never a production consumer group, CDC server id, replication slot, or service account. 3. Bounded reads by default, so a validation run cannot sit on a shared cluster for hours. 4. Local state and output stay under a local scratch directory. 5. Credentials live only in a git-ignored file. Ask the user to name each endpoint and confirm it is development or staging. Never infer it from the hostname, and never reuse an endpoint found in a production manifest. When the only available endpoint is production, stop and say so: generate the data instead (`datagen`, a filesystem fixture, or `VALUES`). Reuse the endpoints and accounts the project already has; do not invent hostnames or create credentials. **Kafka.** Set a dedicated `properties.group.id` (`local-dev-<user>-<job>`) — when the option is omitted the connector generates `KafkaSource-<tableIdentifier>`, which collides across runs. Leave checkpointing off: offsets are committed back to Kafka on checkpoint, and that is the one way a read-only source run can still change shared state. `earliest-offset` on a large topic reads the whole retention window; prefer a timestamp slice. Never declare a Kafka sink locally. **JDBC.** Use an account with `SELECT` only, and verify it — an account that can write is a failed guardrail even if this run happens not to write. Keep the scan small (a `LIMIT`ed query, a partitioned scan, a narrow predicate); a full-table scan on a shared development database is a visible load event. Lookup joins issue per-row queries: set a cache or bound the driving side first. **CDC.** The highest-risk local source, because it registers as a replication client on a shared database. MySQL `server-id` must be unique across every client, and a parallel incremental snapshot needs a range (`'server-id' = '<start>-<end>'`) wide enough for the reader count — a collision with production breaks the production job. Postgres `slot.name` must be unique, and a leftover slot retains WAL indefinitely and can fill the primary's disk, so drop it after the run. Prefer `scan.startup.mode = 'latest-offset'`: the default snapshot reads whole tables. A CDC source never ends — rely on the timeout and treat the run as a sampled observation. **Sinks.** Every `INSERT INTO`/`INSERT OVERWRITE` target gets a `CREATE TEMPORARY TABLE` shadow whose column list matches the production DDL exactly. Allowed local connectors: | Connector | Use it for | Note | |---|---|---| | `print` | inspecting rows and changelog kinds | rows land on client stdout | | `blackhole` | plan, throughput, and state checks | produces no output | | `filesystem` on a `file://` path | keeping output to diff between runs | commits part files only on checkpoint | A temporary table shadows the permanent or script-declared table of the same name and makes it inaccessible for the session, so the job script needs no edit. If the user insists on a real sink, that is no longer a local validation run: require explicit approval, a dedicated scratch topic/table (`local_dev_` prefix), and non-production credentials — then say plainly in the report that a real system was written to. **State paths.** Local checkpoint, savepoint, and output directories live under a scratch path such as `/tmp/datus-flink-local/<name>/`. Never point them at production state: a local job writing there can corrupt what a production job depends on. Validating against production state is copy-then-restore — copy the savepoint locally, restore from the copy. **Credentials.** Flink SQL DDL does not expand environment variables, so a development password in a `WITH` clause is a literal. Keep every overlay file that carries one out of version control (`deploy/flink/*/local/*.local.sql` in `.gitignore`); never write a credential into `sql/job.sql` — production reads secrets from Kubernetes Secrets; never echo an overlay file into the transcript, a log, or a report. Stop and ask instead of guessing when an endpoint's environment is unconfirmed, a sink has no local shadow, the credential offered can write, a CDC `server-id`/`slot.name` cannot be confirmed free, an unbounded source would run without a timeout, or a checkpoint path is not local. ## 4. Assemble the local overlay Three files, loaded in this order through the SQL Client's `-i` option. `-i` takes a single file, so the runner concatenates them; build the same file by hand (`cat local/local-session.sql local/local-sources.sql local/local-sinks.sql > /tmp/local-init.sql`) when invoking the client directly. An init script accepts `SET`, `RESET`, DDL, `USE`, and `LOAD/UNLOAD MODULE` — not queries or inserts. Leave `sql/job.sql` untouched. If a local run only works after editing the job script, that edit is either a real fix (keep it, it belongs in production) or a local workaround (move it to the overlay). ### local-session.sql ```sql -- Local validation session settings for __JOB_NAME__. -- Local only: never copy these into sql/job.sql or spec.flinkConfiguration. -- Required. Runs the job in a MiniCluster inside this client JVM instead of -- submitting it to whatever cluster flink-conf.yaml points at. SET 'execution.target' = 'local'; -- Required. Waits for the INSERT job to finish so the script's exit code -- reflects the job result instead of just the submission. SET 'table.dml-sync' = 'true'; -- Must match production. SET 'execution.runtime-mode' = '__RUNTIME_MODE__'; SET 'table.local-time-zone' = '__TIME_ZONE__'; -- Start at 1 to keep output readable; re-run at the production value once the -- logic is right, to expose ordering and keyed-state assumptions. SET 'parallelism.default' = '__PARALLELISM__'; SET 'sql-client.execution.result-mode' = 'tableau'; -- Off while validating logic: checkpointing adds latency and is what would -- commit Kafka offsets back to a shared consumer group. Enable it only to test -- a filesystem sink, state size, or restore behaviour — always to a local path. -- SET 'execution.checkpointing.interval' = '10s'; -- SET 'state.checkpoints.dir' = 'file://__LOCAL_STATE_DIR__/checkpoints'; -- Flink 2.x renamed the option above to execution.checkpointing.dir. -- Uncomment when a windowed or interval query emits nothing because one -- partition is idle and the watermark never advances. -- SET 'table.exec.source.idle-timeout' = '10s'; ``` ### local-sources.sql ```sql -- Optional local source overlay for __JOB_NAME__. The column list, types, and -- watermark must match the production DDL exactly. Read development endpoints -- only, read-only, bounded. Keep this file git-ignored when it carries a -- credential. -- Bounded slice of a development Kafka topic. CREATE TEMPORARY TABLE __SOURCE_TABLE__ ( __SOURCE_COLUMNS__ ) WITH ( 'connector' = 'kafka', 'topic' = '__DEV_TOPIC__', 'properties.bootstrap.servers' = '__DEV_BOOTSTRAP_SERVERS__', -- Dedicated group: never a production consumer group. 'properties.group.id' = 'local-dev-__LOCAL_RUN_ID__', -- Recent slice, so the run reads real data and still terminates. 'scan.startup.mode' = 'timestamp', 'scan.startup.timestamp-millis' = '__START_TIMESTAMP_MILLIS__', 'scan.bounded.mode' = 'latest-offset', 'format' = 'json', 'json.ignore-parse-errors' = 'false' ); -- Variant: generated data, when no development endpoint may be read. -- CREATE TEMPORARY TABLE __SOURCE_TABLE__ ( -- __SOURCE_COLUMNS__ -- ) WITH ( -- 'connector' = 'datagen', -- -- A sequence field bounds the table; 'number-of-rows' bounds a random one. -- 'fields.__SOURCE_KEY_COLUMN__.kind' = 'sequence', -- 'fields.__SOURCE_KEY_COLUMN__.start' = '1', -- 'fields.__SOURCE_KEY_COLUMN__.end' = '20' -- ); -- Variant: fixture files checked in next to the job, fully reproducible. -- CREATE TEMPORARY TABLE __SOURCE_TABLE__ ( -- __SOURCE_COLUMNS__ -- ) WITH ( -- 'connector' = 'filesystem', -- 'path' = 'file://__LOCAL_FIXTURE_DIR__/__SOURCE_TABLE__', -- 'format' = 'json' -- ); -- Variant: read-only development database. Keep the scan small, use an account -- that cannot write, and .gitignore this file before adding a password. -- CREATE TEMPORARY TABLE __SOURCE_TABLE__ ( -- __SOURCE_COLUMNS__ -- ) WITH ( -- 'connector' = 'jdbc', -- 'url' = 'jdbc:mysql://__DEV_DB_HOST__:3306/__DEV_DB__', -- 'table-name' = '__DEV_TABLE__', -- 'username' = '__DEV_READONLY_USER__', -- 'password' = '__DEV_READONLY_PASSWORD__' -- ); ``` ### local-sinks.sql ```sql -- Required local sink overlay for __JOB_NAME__. Every INSERT INTO / -- INSERT OVERWRITE target in the job script needs a shadow here, with the -- production column list. Only print, blackhole, and filesystem on a file:// -- path are allowed: a local run never writes to a real topic, table, or object -- store. CREATE TEMPORARY TABLE __SINK_TABLE__ ( __SINK_COLUMNS__ ) WITH ( 'connector' = 'print', 'print-identifier' = '__SINK_TABLE__' ); -- Variant: keep the output as files to diff between runs. The filesystem sink -- commits part files only on checkpoint, so enable -- execution.checkpointing.interval in local-session.sql when using this. -- CREATE TEMPORARY TABLE __SINK_TABLE__ ( -- __SINK_COLUMNS__ -- ) WITH ( -- 'connector' = 'filesystem', -- 'path' = 'file://__LOCAL_OUTPUT_DIR__/__SINK_TABLE__', -- 'format' = 'json' -- ); -- Variant: discard the rows, for plan, state, and throughput checks only. -- CREATE TEMPORARY TABLE __SINK_TABLE__ ( -- __SINK_COLUMNS__ -- ) WITH ( -- 'connector' = 'blackhole' -- ); ``` ## 5. Run it through the guarded runner Write this script into `local/run-local-sql.sh` and run the job through it. It fails closed: no unpinned session, no unshadowed sink, no non-local sink, no tracked credential. Never bypass a guard by calling `sql-client.sh` directly with a real endpoint configured. ### run-local-sql.sh ```bash #!/usr/bin/env bash # # Run a Flink SQL job in an in-process MiniCluster for local validation. # Exit codes: 0 job ok · 2 usage · 3 environment · 4 guard · 5 timeout · # anything else is the SQL Client's own exit code. set -euo pipefail JOBS=(); JARS=(); SESSION=""; SOURCES=""; SINKS=""; TIMEOUT_SECONDS=600; EXPECT="" USAGE="usage: run-local-sql.sh --job <file.sql>... --session <local-session.sql>