apache/flink · error · ValidationException

Could not find any format factory for identifier '%s' in the

Error message

Could not find any format factory for identifier '%s' in the classpath.

What it means

FileSystemTableSink's constructor checks that at least one writer format (bulkWriterFormat or serializationFormat) is non-null. If both are null — meaning no encoding format was discovered for writing — it throws a ValidationException. The format identifier is retrieved from table options for the error message. This is a table-creation-time validation performed in the constructor.

Source

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

    private LinkedHashMap<String, String> staticPartitions = new LinkedHashMap<>();

    @Nullable private Integer configuredParallelism;

    FileSystemTableSink(
            ObjectIdentifier tableIdentifier,
            DataType physicalRowDataType,
            List<String> partitionKeys,
            ReadableConfig tableOptions,
            @Nullable DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat,
            @Nullable DecodingFormat<DeserializationSchema<RowData>> deserializationFormat,
            @Nullable EncodingFormat<BulkWriter.Factory<RowData>> bulkWriterFormat,
            @Nullable EncodingFormat<SerializationSchema<RowData>> serializationFormat) {
        super(tableIdentifier, physicalRowDataType, partitionKeys, tableOptions);
        this.bulkReaderFormat = bulkReaderFormat;
        this.deserializationFormat = deserializationFormat;
        if (Stream.of(bulkWriterFormat, serializationFormat).allMatch(Objects::isNull)) {
            String identifier = tableOptions.get(FactoryUtil.FORMAT);
            throw new ValidationException(
                    String.format(
                            "Could not find any format factory for identifier '%s' in the classpath.",
                            identifier));
        }
        this.bulkWriterFormat = bulkWriterFormat;
        this.serializationFormat = serializationFormat;
        this.configuredParallelism =
                this.tableOptions.get(FileSystemConnectorOptions.SINK_PARALLELISM);
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context sinkContext) {
        return new DataStreamSinkProvider() {
            @Override
            public DataStreamSink<?> consumeDataStream(
                    ProviderContext providerContext, DataStream<RowData> dataStream) {
                return consume(providerContext, dataStream, sinkContext);
            }

View on GitHub (pinned to 2f3c205e92)

Solutions

Ensure the format JAR (e.g. flink-sql-parquet, flink-sql-csv) is on the classpath of the SQL client or job submission.
Verify the format identifier supports writing (most common formats do: parquet, csv, json, orc, avro).
Check for typos in the 'format' option value.
If the format only supports reading, use a different format for the sink or restructure the pipeline.

Example fix

-- before
CREATE TABLE sink_t (a INT) WITH (
  'connector'='filesystem',
  'path'='file:///out',
  'format'='unknownformat'
);

-- after
CREATE TABLE sink_t (a INT) WITH (
  'connector'='filesystem',
  'path'='file:///out',
  'format'='parquet'
);
-- ensure flink-sql-parquet JAR is on the classpath

Defensive patterns

Strategy: validation

Validate before calling

// Verify format JAR availability before table creation
String formatId = options.get("format");
if (formatId == null) {
    throw new ValidationException("'format' option is required for filesystem sink");
}
// Check ServiceLoader for matching EncodingFormat factory
boolean found = ServiceLoader.load(Factory.class)
    .stream()
    .anyMatch(p -> p.get().identifier().equals(formatId));
if (!found) {
    throw new ValidationException("No format factory for '" + formatId + "'. Add the format JAR.");
}

Prevention

Ensure the format JAR (flink-sql-parquet, flink-sql-csv, etc.) is on the classpath.
Verify the format identifier supports writing, not just reading.
Use a consistent set of format JARs across the cluster.
Test table creation in development before production deployment.

When it happens

Trigger: A CREATE TABLE with connector='filesystem' specifies a 'format' option value for which no EncodingFormat factory is found in the classpath. For example, specifying 'format'='xyz' where 'xyz' has a reader factory but no writer factory, or specifying a format whose JAR is not on the classpath.

Common situations: The format JAR (e.g. flink-parquet, flink-csv) is not on the classpath at table creation time. The specified format supports reading but not writing. Typo in the format identifier. Using a read-only format for a sink table.

Related errors

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