apache/flink · error · ValidationException

Currently, filesystem sink doesn't support setting paralleli

Error message

Currently, filesystem sink doesn't support setting parallelism (%d) by '%s' when the input stream is not INSERT only. The row kinds of input stream are [%s]

What it means

FileSystemTableSink.checkConfiguredParallelismAllowed validates that if sink.parallelism is configured, the input stream must be INSERT-only (no UPDATE_BEFORE, UPDATE_AFTER, or DELETE row kinds). Filesystem sinks with explicit parallelism distribute rows across subtasks, and non-INSERT changelog rows would be split across files, breaking correctness. If the changelog contains non-INSERT kinds, a ValidationException is thrown listing the parallelism value, the option key, and the row kinds found.

Source

Thrown at flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java:407

        if (bulkWriterFormat != null) {
            return bulkWriterFormat.createRuntimeEncoder(
                    sinkContext, physicalDataTypeWithoutPartitionColumns);
        } else if (serializationFormat != null) {
            return new SerializationSchemaAdapter(
                    serializationFormat.createRuntimeEncoder(
                            sinkContext, physicalDataTypeWithoutPartitionColumns));
        } else {
            throw new TableException("Can not find format factory.");
        }
    }

    private void checkConfiguredParallelismAllowed(ChangelogMode requestChangelogMode) {
        final Integer parallelism = this.configuredParallelism;
        if (parallelism == null) {
            return;
        }
        if (!requestChangelogMode.containsOnly(RowKind.INSERT)) {
            throw new ValidationException(
                    String.format(
                            "Currently, filesystem sink doesn't support setting parallelism (%d) by '%s' "
                                    + "when the input stream is not INSERT only. The row kinds of input stream are [%s]",
                            parallelism,
                            FileSystemConnectorOptions.SINK_PARALLELISM.key(),
                            requestChangelogMode.getContainedKinds().stream()
                                    .map(RowKind::shortString)
                                    .collect(Collectors.joining(","))));
        }
    }

    private static OutputFormat<RowData> createBulkWriterOutputFormat(
            BulkWriter.Factory<RowData> factory, Path path) {
        return new OutputFormat<RowData>() {

            private static final long serialVersionUID = 1L;

            private transient BulkWriter<RowData> writer;

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Remove the sink.parallelism option — let the filesystem sink use the input parallelism (changelog mode is then implicitly handled).
2. If parallelism control is essential, ensure the upstream query only produces INSERT-only rows (e.g. no aggregations or updates before the sink).
3. Insert a deduplication or append-only materialization step before the filesystem sink to normalize the changelog to INSERT-only.

Example fix

-- before
CREATE TABLE sink_t (a INT, cnt BIGINT) WITH (
  'connector'='filesystem',
  'path'='file:///out',
  'format'='parquet',
  'sink.parallelism'='4'
);
INSERT INTO sink_t SELECT a, COUNT(*) FROM source GROUP BY a;
-- aggregation produces UPDATE rows -> ValidationException

-- after
-- remove sink.parallelism:
CREATE TABLE sink_t (a INT, cnt BIGINT) WITH (
  'connector'='filesystem',
  'path'='file:///out',
  'format'='parquet'
);
INSERT INTO sink_t SELECT a, COUNT(*) FROM source GROUP BY a;

// Before setting sink.parallelism, verify changelog mode is INSERT-only
if (configuredParallelism != null && !changelogMode.containsOnly(RowKind.INSERT)) {
    throw new ValidationException(
        "sink.parallelism is not supported for non-INSERT-only streams. "
        + "Remove 'sink.parallelism' or ensure INSERT-only input.");
}

When it happens

Trigger: The table option sink.parallelism is set to a non-null value AND the input changelog mode contains row kinds other than INSERT. This happens when the upstream query produces an upsert or changelog stream (e.g. aggregations, joins that produce updates) that is written to a filesystem sink with explicit parallelism.

Common situations: Writing the output of a GROUP BY aggregation (which produces UPDATE rows) to a filesystem sink with sink.parallelism set. Upsert Kafka source feeding a filesystem sink with custom parallelism. Any changelog-producing operator upstream of a filesystem sink with parallelism configured.

Related errors

• Could not find any format factory for identifier '%s' in the
• TaskNumber: {}, numTasks: {}
• Table options do not contain an option key '%s' for discover
• Invalid time zone for '%s'. The value should be a Time Zone
• Streaming mode not support overwrite.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/01279c68c5acfc58. Report an issue: GitHub.