Run Batch Statements with Confluent Manager for Apache Flink

A batch statement runs its SQL query as an Apache Flink® batch job. The job reads a bounded amount of input, writes its result and finishes. Use batch statements for work that has an end, such as a backfill into a table or a report over a fixed range of a topic. A statement runs in batch mode when execution.runtime-mode is BATCH, set either in the statement’s flinkConfiguration or in the environment’s statement defaults. When both set it, the environment’s value wins, as described in How CMF decides the runtime mode.

Batch statements have the following requirements:

The statement must run on a compute pool that runs Flink 2.x. On a compute pool that runs Flink 1.x, such as 1.19, CMF rejects the create request with a 400 and the message batch statements require Flink 2.x.
Every table the statement reads must be bounded. CMF accepts a batch statement that reads an unbounded source, but the statement moves to FAILED with an error that says the query requires all referenced sources to be bounded.
execution.runtime-mode must be BATCH or STREAMING. CMF rejects AUTOMATIC with a 400, in a statement and in an environment’s statement defaults.

Create a batch statement

A table backed by a Apache Kafka® topic is unbounded by default. To read it as a bounded source, set the scan.bounded.mode table option, either with CREATE TABLE or ALTER TABLE, or in a query hint as in the following example. With latest-offset, the job reads each partition up to the offset that was the latest when the job started. For the other values, see Table options.

scan.bounded.mode sets only where the job stops reading. It starts reading where the table’s scan.startup.mode says. To read a topic from its first offset, whatever the table sets, also set scan.startup.mode to earliest-offset in the hint.

The following statement copies all orders that exist when the job starts into the orders_archive table:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Statement",
  "metadata": {
    "name": "archive-orders"
  },
  "spec": {
    "statement": "INSERT INTO orders_archive SELECT * FROM orders /*+ OPTIONS('scan.startup.mode'='earliest-offset', 'scan.bounded.mode'='latest-offset') */;",
    "properties": {
      "sql.current-catalog": "kafka-cat",
      "sql.current-database": "kafka-db"
    },
    "flinkConfiguration": {
      "execution.runtime-mode": "BATCH"
    },
    "computePoolName": "pool-flink-2"
  }
}

Create the statement as described in Create Statements with Confluent Manager for Apache Flink. A batch INSERT INTO statement or statement set does not need execution.checkpointing.interval, because batch jobs do not checkpoint.

After CMF compiles the statement, status.traits.runtimeMode shows the mode its job runs in, BATCH or STREAMING. Check this field rather than the statement’s own flinkConfiguration, because the environment can override it, as described in the next section.

When the job finishes, the statement moves to COMPLETED. For an INSERT INTO statement or statement set, CMF then deletes the job’s Flink resources, unless you turned off INSERT statement resource cleanup. A batch SELECT statement moves to COMPLETED when you read the last page of its results. To run any batch statement again, create a new statement.

A batch SELECT statement buffers its results until you fetch them. The buffer holds 4MiB by default, and when it is full, the job waits. For a result larger than that, fetch the results while the statement runs, not after it finishes. For details, see Manage Statements in Confluent Manager for Apache Flink.

How CMF decides the runtime mode

A statement’s runtime mode can come from two places:

The flinkConfiguration of the statement.
The environment’s statement defaults. statementDefaults.interactive applies to SELECT statements, and statementDefaults.detached applies to INSERT INTO statements and statement sets.

When both set execution.runtime-mode, the environment’s statement defaults win. When neither sets it, the statement runs in streaming mode. The compute pool’s Flink configuration and the environment’s computePoolDefaults do not count: execution.runtime-mode set there has no effect on statements. A SET 'execution.runtime-mode' = 'batch' statement fails with Statement type not supported yet., so set the runtime mode in one of the two places above.

This means that you can make every INSERT INTO statement and statement set in an environment a batch statement by setting the runtime mode in the detached statement defaults:

"statementDefaults": {
  "detached": {
    "flinkConfiguration": {
      "execution.runtime-mode": "BATCH"
    }
  }
}

It also means that a statement that sets BATCH itself runs in streaming mode when the environment’s defaults set STREAMING. This holds whether the statement’s sources are bounded or not. A streaming INSERT INTO statement or statement set needs a checkpoint interval. Without one, CMF rejects the create request with a 400 and an error that says the environment’s defaults take precedence over the statement’s BATCH.

CMF compiles a statement for the runtime mode it has when you create it, and the statement keeps that mode. If you change execution.runtime-mode in an environment’s statement defaults, CMF rejects the update with a 400 while any statement in the environment would switch between batch and streaming mode, and the error names those statements. Stopped statements block the update too. Statements that are COMPLETED or FAILED do not. Delete the statements the error names, and then update the environment.

How batch statements differ from streaming statements

Savepoints

Flink cannot take a savepoint of a batch job or start a batch job from one. CMF therefore rejects savepoint settings on batch statements:

Creating a batch statement with spec.savepointSchedule fails with a 400. An update that adds a schedule fails with a 422.
Creating or updating an environment whose statementDefaults.detached sets both BATCH and a savepointSchedule fails with a 400.
A statement that sets BATCH in its own flinkConfiguration does not inherit a savepoint schedule from the environment’s detached statement defaults.
A request to create a savepoint for a batch statement fails with a 400.
An update that adds spec.startFromSavepoint to a batch statement fails with a 422.

Stop, resume, and rescale

Stopping a batch statement cancels its job immediately. A batch job cannot continue from where it stopped, so starting it again would run the whole query from the beginning and could write its output a second time. For a batch INSERT INTO statement or statement set, CMF prevents this:

Resuming the statement after its job has run fails with a 400.
An update that would restart the running job, such as a change to parallelism, fails with a 400. An update that does not touch the job, such as a label change, is accepted. If the environment’s defaults or the compute pool’s configuration changed since the job started, or CMF was upgraded or reconfigured, applying any update would restart the job, so a label change is rejected too.
Stopping the statement is always accepted.

A statement that you create with "stopped": true has not run a job yet, so you can start it once. To run the query again after its job has run, create a new statement with another name, or delete the old statement first.

Task failures

Flink restarts a failed task only when a restart strategy is configured or checkpointing is on, and batch jobs do not checkpoint. CMF therefore adds a default restart strategy to batch INSERT INTO statements and statement sets: exponential-delay with 10 attempts, which retries for about two minutes before the statement fails. Operators can change this default, as described in Configure the default restart strategy.

When Flink restarts the tasks of a batch job, the restarted tasks write their output again. A sink that is not transactional, such as an at-least-once Kafka sink, can then receive duplicate rows.

A batch SELECT statement gets no default restart strategy. Unless you configure one, its job fails on the first task failure, such as a lost TaskManager.

Known limitations

The following problems are known in CMF 2.5. Each has a workaround.

Bounding a source in a window function

Flink does not accept a table hint on the TABLE argument of a window function such as TUMBLE. The statement fails with SQL parse failed. Encountered "/*+". Instead, create a view in the environment catalog that reads the table with the hint, and run the window function over the view:

CREATE VIEW orders_bounded AS
  SELECT * FROM `kafka-cat`.`kafka-db`.`orders`
  /*+ OPTIONS('scan.startup.mode'='earliest-offset',
             'scan.bounded.mode'='latest-offset') */;

SELECT window_start, window_end, COUNT(*)
  FROM TABLE(
    TUMBLE(TABLE orders_bounded, DESCRIPTOR(ts), INTERVAL '10' SECOND))
  GROUP BY window_start, window_end;

SELECT * does not include the $rowtime column of a Kafka table. If the window uses $rowtime, add it to the view’s select list.

You can also set scan.bounded.mode on the table itself with ALTER TABLE, but then every statement that reads the table, including streaming statements, reads it as a bounded source.

Bounded-mode options are not checked

CMF does not check the scan.bounded.* options. A missing or wrong option does not fail the statement at create:

'scan.bounded.mode'='timestamp' without scan.bounded.timestamp-millis reads no rows, and the statement completes with an empty result.
'scan.bounded.mode'='specific-offsets' with an offset beyond the end of a partition never finishes. Without scan.bounded.specific-offsets, the job fails after it starts.
'scan.bounded.mode'='group-offsets' without properties.group.id fails after the job starts.
A misspelled option name fails with the error that the query requires bounded sources, and does not name the option.

Use latest-offset unless you need another end point, and set the option that each mode requires, as described in Table options.

Parallelism below the compute pool’s minimum

If the compute pool sets execution.batch.adaptive.auto-parallelism.min-parallelism higher than the statement’s parallelism, which defaults to 1, the job fails at start with Maximum parallelism should be greater than or equal to the minimum parallelism. Set the statement’s parallelism to at least the pool’s minimum.

Joins followed by GROUP BY on the Default scheduler

On a compute pool that sets jobmanager.scheduler: default, a batch join followed by a GROUP BY on the join key fails at job start with Could not start the JobMaster. Do not set jobmanager.scheduler on compute pools that run batch statements, so that Flink uses its default adaptive batch scheduler.

Runtime filters

With table.optimizer.runtime-filter.enabled: true, a batch join fails to compile, and the statement fails with Unable to serialize given ExecNodeGraph. Do not set this option.

The NEST_LOOP join hint

A batch query with a /*+ NEST_LOOP(...) */ hint compiles, but its job fails. Leave the join strategy to the planner, or use the SHUFFLE_HASH or SHUFFLE_MERGE hint.

Batch job recovery without high availability

If a compute pool sets execution.batch.job-recovery.enabled: true but no high-availability.storageDir, its JobManager fails at start, and every statement on the pool fails. The error appears only in the JobManager log. Set high-availability.storageDir on any pool that enables batch job recovery.

Reading Iceberg tables

A batch statement that reads an Iceberg table can start many TaskManagers for a small query, and can fail with Insufficient number of network buffers. Add table.exec.iceberg.infer-source-parallelism: false to the statement’s flinkConfiguration.

Configure the default restart strategy

Operators set the restart strategy that CMF adds to batch INSERT INTO statements and statement sets with the cmf.sql.batchRestartStrategy Helm value.

Helm value	Default	Description
`cmf.sql.batchRestartStrategy.enabled`	`true`	Set to `false` to add no restart strategy.
`cmf.sql.batchRestartStrategy.flinkConfiguration`	`exponential-delay` with 10 attempts	The Flink restart strategy options to add. Only `restart-strategy.*` options are allowed, and `restart-strategy.type` is required. The default sets `restart-strategy.type` to `exponential-delay` and `restart-strategy.exponential-delay.attempts-before-reset-backoff` to `"10"`.

CMF adds these options only when none of the following sets restart-strategy.type: the statement, the environment’s statement defaults, the compute pool, and the environment’s compute pool defaults. A restart strategy that you set in any of these places is kept. CMF does not store the options in the statement, so they do not appear in the statement’s flinkConfiguration. They appear in the job configuration in the Flink web UI.

To use a fixed delay between restarts instead, follow these steps:

Add the following to your Helm values file. Quote numbers, because the values must be strings.

cmf:
  sql:
    batchRestartStrategy:
      enabled: true
      flinkConfiguration:
        restart-strategy.type: fixed-delay
        restart-strategy.fixed-delay.attempts: "5"
        restart-strategy.fixed-delay.delay: "30 s"

Upgrade CMF:

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
  --version "~2.5.0" \
  --namespace <namespace> \
  -f values.yaml

Helm merges your flinkConfiguration with the default, so the default’s restart-strategy.exponential-delay.attempts-before-reset-backoff option is still added. Flink ignores it when restart-strategy.type is not exponential-delay.

The chart’s schema rejects most invalid values when you run helm upgrade: options that are not Flink restart strategy options, values that are not strings, and a missing restart-strategy.type. If a value passes the schema but Flink rejects it, CMF does not start, and its log names the option.

The change applies to batch statements that start after the upgrade. A batch job that is already running finishes with the restart strategy it started with.