{"record":{"id":"ef5acca53a3d6a2c","repo":"apache/flink","slug":"could-not-find-any-format-factory-for-identifier","errorCode":null,"errorMessage":"Could not find any format factory for identifier '%s' in the classpath.","messagePattern":"Could not find any format factory for identifier '(.+?)' in the classpath\\.","errorType":"validation","errorClass":"ValidationException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java","lineNumber":126,"sourceCode":"    private LinkedHashMap<String, String> staticPartitions = new LinkedHashMap<>();\n\n    @Nullable private Integer configuredParallelism;\n\n    FileSystemTableSink(\n            ObjectIdentifier tableIdentifier,\n            DataType physicalRowDataType,\n            List<String> partitionKeys,\n            ReadableConfig tableOptions,\n            @Nullable DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat,\n            @Nullable DecodingFormat<DeserializationSchema<RowData>> deserializationFormat,\n            @Nullable EncodingFormat<BulkWriter.Factory<RowData>> bulkWriterFormat,\n            @Nullable EncodingFormat<SerializationSchema<RowData>> serializationFormat) {\n        super(tableIdentifier, physicalRowDataType, partitionKeys, tableOptions);\n        this.bulkReaderFormat = bulkReaderFormat;\n        this.deserializationFormat = deserializationFormat;\n        if (Stream.of(bulkWriterFormat, serializationFormat).allMatch(Objects::isNull)) {\n            String identifier = tableOptions.get(FactoryUtil.FORMAT);\n            throw new ValidationException(\n                    String.format(\n                            \"Could not find any format factory for identifier '%s' in the classpath.\",\n                            identifier));\n        }\n        this.bulkWriterFormat = bulkWriterFormat;\n        this.serializationFormat = serializationFormat;\n        this.configuredParallelism =\n                this.tableOptions.get(FileSystemConnectorOptions.SINK_PARALLELISM);\n    }\n\n    @Override\n    public SinkRuntimeProvider getSinkRuntimeProvider(Context sinkContext) {\n        return new DataStreamSinkProvider() {\n            @Override\n            public DataStreamSink<?> consumeDataStream(\n                    ProviderContext providerContext, DataStream<RowData> dataStream) {\n                return consume(providerContext, dataStream, sinkContext);\n            }","sourceCodeStart":108,"sourceCodeEnd":144,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java#L108-L144","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"-- before\nCREATE TABLE sink_t (a INT) WITH (\n  'connector'='filesystem',\n  'path'='file:///out',\n  'format'='unknownformat'\n);\n\n-- after\nCREATE TABLE sink_t (a INT) WITH (\n  'connector'='filesystem',\n  'path'='file:///out',\n  'format'='parquet'\n);\n-- ensure flink-sql-parquet JAR is on the classpath","handlingStrategy":"validation","validationCode":"// Verify format JAR availability before table creation\nString formatId = options.get(\"format\");\nif (formatId == null) {\n    throw new ValidationException(\"'format' option is required for filesystem sink\");\n}\n// Check ServiceLoader for matching EncodingFormat factory\nboolean found = ServiceLoader.load(Factory.class)\n    .stream()\n    .anyMatch(p -> p.get().identifier().equals(formatId));\nif (!found) {\n    throw new ValidationException(\"No format factory for '\" + formatId + \"'. Add the format JAR.\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["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."],"tags":["table-sink","format-factory","validation","classloader","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}