apache-flink-knowledge-patch

# Apache Flink Knowledge Patch Use this skill when upgrading or operating Flink, writing DataStream or Table API jobs, changing SQL, implementing connectors, or diagnosing state, checkpoint, scheduling, and deployment behavior. Start with the quick checks, then open the topic reference that matches the work. ## Reference index | Reference | Topics | | --- | --- | | [Migration, APIs, and Configuration](references/migration-apis-and-configuration.md) | Java/runtime requirements, removed APIs, typed configuration, SPI changes, serialization compatibility | | [SQL, Table API, and Planning](references/sql-table-and-planning.md) | SQL syntax and types, materialized tables, inference, joins, PTFs, compiled plans, planner corrections | | [State, Checkpoints, and Storage](references/state-checkpoints-and-storage.md) | Async/disaggregated state, savepoints, RocksDB, ForSt, unaligned checkpoints, native S3 | | [Runtime, Scheduling, and Deployment](references/runtime-scheduling-and-deployment.md) | Adaptive Scheduler, partitioning, watermarks, Kubernetes, YARN, applications, REST and Web UI | | [Connectors, Formats, and Python](references/connectors-formats-and-python.md) | Source/Sink V2, connector migration, batching, formats, PyFlink, filesystems and HTTP behavior | | [Observability, Security, and Operations](references/observability-security-and-operations.md) | Metrics, traces, events, HistoryServer retention, logging, dependencies, TLS and operational fixes | ## Upgrade triage Before changing a job or cluster: 1. Identify the exact Flink line and every external connector version. 2. Inventory removed DataSet, Scala, Source/Sink V1, legacy Table, state, serialization, deployment, and configuration surfaces. 3. Treat state and savepoint restoration across the 1.x/2.x boundary as a compatibility project, not an assumed in-place restore. 4. Convert `flink-conf.yaml` to standard-YAML `config.yaml` before starting a 2.x distribution. 5. Validate the Java runtime: Java 11 is the minimum, Java 17 is the default and recommended runtime, and Java 21 is supported. 6. Rehearse checkpoints, failover, rescaling, and sink commits with the target patch release and production serializers. ## Highest-impact removals Flink 2.0 removes these major families: - Java DataSet and Scala DataStream/DataSet APIs. Prefer Java DataStream or Table API/SQL. - `SourceFunction`, `SinkFunction`, Sink V1, `TableSource`, `TableSink`, `TableSchema`, `TableColumn`, and `Types`. Move to Source/Sink V2, dynamic table sources/sinks, `Schema`, `Column`, and `DataTypes`. - Positional/string `keyBy` and `partitionCustom`, iterations, legacy watermark assignment, convenience file writers/readers, and `DataStreamUtils.collect*`. - `ParameterTool` and its abstract/multiple variants. - Explicit `TimeCharacteristic` selection and old Flink `Time` overloads. - Per-job deployment and legacy Hybrid Shuffle. Use application mode; submit Kubernetes applications with `flink run -t kubernetes-application`. - The SQL Client `-u/--update` option. Put updates in a file and use `-f`. Open the migration reference before resolving compilation errors: several replacement APIs change lifecycle contexts or return types rather than simply renaming a method. ## Configuration quick reference Use typed options: ```java configuration.set(option, value); T value = configuration.get(option); ``` `RichFunction.open(Configuration)` becomes `open(OpenContext)`. UDFs obtain serializer, global-parameter, and object-reuse facilities from `RuntimeContext`; they no longer receive the full `ExecutionConfig` surface. For Adaptive Scheduler local recovery, opt in explicitly: ```yaml execution.state-recovery.from-local: true ``` On Kubernetes, disable minimal-TaskManager preference only when the prior downscaling strategy is required: ```yaml jobmanager.adaptive-scheduler.prefer-minimal-taskmanagers: false ``` Do not copy removed 1.x options into 2.x configuration. In particular, review state-backend, Kryo/POJO registration, SSL, heap, web/backpressure, network, adaptive-batch, and Hybrid Shuffle settings against the typed 2.x options. ## SQL and Table quick reference Use `QUALIFY` to filter window-function results: ```sql SELECT * FROM orders QUALIFY ROW_NUMBER() OVER ( PARTITION BY customer_id ORDER BY order_time DESC ) = 1; ``` Table functions may appear directly in `FROM`; the outer `TABLE(...)` wrapper remains valid. SQL also supports C-style escape strings and casts from `DOUBLE` to `BOOLEAN`. For semi-structured data, `VARIANT` preserves scalar, array, and string-keyed map types. Convert JSON text with `PARSE_JSON` or `TRY_PARSE_JSON`. For catalog-defined real-time inference, use `ML_PREDICT`; Table API can also define models and run inference. `VECTOR_SEARCH` adds online vector similarity search. Keep provider-specific availability separate from the core API. Materialized tables support declarative refresh pipelines, cross-cluster YARN and Kubernetes submission, Paimon integration, schema/query evolution, bucketing, discovery, explicit columns, watermarks, primary keys, and selectable refresh start positions. A physical-plan replacement can still require full reprocessing. ## Join and process-function cautions The multi-join operator is experimental. It is opt-in: ```sql SET 'table.optimizer.multi-join.enabled' = 'true'; ``` Its state keys changed from upsert keys to unique keys after its initial release. Use a patch release containing the relevant OR-predicate, heap-backend, mixed-source, and row-kind fixes before relying on it. Process Table Functions support managed state, timers, event time, changelogs, late-record handling, and ordered table arguments: ```sql MyPtf(input => TABLE t PARTITION BY k ORDER BY ts) ``` When an upsert key differs from the sink primary key, choose an explicit `ON CONFLICT` action: `DO NOTHING`, `DO ERROR`, or `DO DEDUPLICATE`. ## State and checkpoint quick reference Flink's disaggregated state architecture uses remote primary storage and parallel asynchronous access. DataStream `process()` supports async state, and DataStream V2 state lookup returns V2 state objects directly rather than legacy state wrapped in `Optional`. Sink expansion disables unaligned checkpoints only within expanded pre-commit, committer, or post-commit connections. A broad disablement on all connections was a regression corrected in 2.0.1. To checkpoint progress while restored channel state is still being consumed, enable both controls: ```yaml execution.checkpointing.unaligned.recover-output-on-downstream.enabled: true execution.checkpointing.unaligned.during-recovery.enabled: true ``` This recovery mode is off by default. Also validate custom partitioners, rescaling, multiple exchanges, and file-merged checkpoints against the patch fixes in the state reference. ## Native S3 quick reference The experimental `flink-s3-fs-native` plugin uses AWS SDK v2, needs neither Hadoop nor Presto, and registers `s3://` and `s3a://`. Install it in the plugins directory. It provides `FileSystem` and `RecoverableWriter` support, including exactly-once streaming sinks and modern role-based authentication. Configure it through `s3.*`; review region, endpoint, path style, credentials, multipart upload, bulk copy, async I/O, read buffers, entropy, SSE-KMS, chunked encoding, and checksum validation. Its experimental status matters even though the implementation is functionally complete. ## Scheduling and watermark quick reference `jobmanager.adaptive-scheduler.executing.resource-stabilization-timeout` delays scaling after resource changes. The old minimum-parallelism-increase control is deprecated. The adaptive network partitioner is disabled by default. Enable it with `taskmanager.network.adaptive-partitioner.enabled`; its channel search limit is `taskmanager.network.adaptive-partitioner.max-traverse-size`, default `4`. Adaptive selection applies to rebalance and rescale partitioners. Watermark idleness excludes time blocked by backpressure or alignment. Custom generators can read the same clock from `WatermarkGeneratorSupplier.Context#getInputActivityClock()`. Watermark alignment waits three update intervals by default before pausing. Set `pipeline.watermark-alignment.buffer-size: 0` only to recover the earlier pause behavior. ## Applications and observability quick reference Applications are lifecycle resources with asynchronous REST submission, overview/detail/cancel endpoints, a Web UI, multi-job membership, failure reporting, and HA re-execution for incomplete session applications. Review the termination, cluster-ID, and HistoryServer application-retention defaults before enabling this lifecycle. Adaptive Scheduler rescale history is disabled at its default size `0`. Set `web.adaptive-scheduler.rescale-history.size` above zero to populate the Web UI **Rescales** tab and rescale REST resources. For OpenTelemetry gRPC metrics, `metrics.reporter.otel.exporter.compression` supports `gzip` instead of default `none`; `metrics.reporter.otel.batch.size` defaults to `0`, meaning batching is disabled. ## Patch-release discipline Do not treat a feature-level upgrade as sufficient when a listed patch release fixes correctness, recovery, security, planner, connector, or deployment behavior relevant to the job. Match symptoms to the topic references and test on the newest compatible patch release before adding workarounds.