{"record":{"id":"9a9a57c8d22068c5","repo":"apache/flink","slug":"can-not-find-format-factory","errorCode":null,"errorMessage":"Can not find format factory.","messagePattern":"Can not find format factory\\.","errorType":"exception","errorClass":"TableException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java","lineNumber":397,"sourceCode":"        return writer instanceof Encoder\n                ? path -> createEncoderOutputFormat((Encoder<RowData>) writer, path)\n                : path -> createBulkWriterOutputFormat((BulkWriter.Factory<RowData>) writer, path);\n    }\n\n    private Object createWriter(Context sinkContext) {\n        DataType physicalDataTypeWithoutPartitionColumns =\n                DataType.getFields(physicalRowDataType).stream()\n                        .filter(field -> !partitionKeys.contains(field.getName()))\n                        .collect(Collectors.collectingAndThen(Collectors.toList(), DataTypes::ROW));\n        if (bulkWriterFormat != null) {\n            return bulkWriterFormat.createRuntimeEncoder(\n                    sinkContext, physicalDataTypeWithoutPartitionColumns);\n        } else if (serializationFormat != null) {\n            return new SerializationSchemaAdapter(\n                    serializationFormat.createRuntimeEncoder(\n                            sinkContext, physicalDataTypeWithoutPartitionColumns));\n        } else {\n            throw new TableException(\"Can not find format factory.\");\n        }\n    }\n\n    private void checkConfiguredParallelismAllowed(ChangelogMode requestChangelogMode) {\n        final Integer parallelism = this.configuredParallelism;\n        if (parallelism == null) {\n            return;\n        }\n        if (!requestChangelogMode.containsOnly(RowKind.INSERT)) {\n            throw new ValidationException(\n                    String.format(\n                            \"Currently, filesystem sink doesn't support setting parallelism (%d) by '%s' \"\n                                    + \"when the input stream is not INSERT only. The row kinds of input stream are [%s]\",\n                            parallelism,\n                            FileSystemConnectorOptions.SINK_PARALLELISM.key(),\n                            requestChangelogMode.getContainedKinds().stream()\n                                    .map(RowKind::shortString)\n                                    .collect(Collectors.joining(\",\"))));","sourceCodeStart":379,"sourceCodeEnd":415,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java#L379-L415","documentation":"FileSystemTableSink.createWriter selects the encoding format for writing. If both bulkWriterFormat and serializationFormat are null (no format was discovered), it throws a TableException 'Can not find format factory.' Unlike the constructor validation (error 193) which throws ValidationException, this is a runtime check reached during getSinkRuntimeProvider when the writer is being created.","triggerScenarios":"createWriter is called when both bulkWriterFormat and serializationFormat fields are null. This should not happen if the constructor validation (error 193) passed, but can occur if the sink was constructed directly (bypassing the factory) or if the formats were nulled after construction (e.g. via copy()).","commonSituations":"Direct construction of FileSystemTableSink without providing a format. A bug in copy() that drops the format references. Factory discovery logic changed in a way that passes null formats to the constructor without triggering the constructor check.","solutions":["Always create FileSystemTableSink through the FileSystemTableFactory to ensure format discovery is performed.","If constructing directly, ensure at least one of bulkWriterFormat or serializationFormat is non-null.","Check copy() implementations to ensure format references are preserved."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Validate formats before constructing the sink\nif (bulkWriterFormat == null && serializationFormat == null) {\n    throw new TableException(\"At least one writer format must be provided\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always create FileSystemTableSink through FileSystemTableFactory.","If constructing directly, provide a non-null format argument.","Ensure copy() implementations preserve all format fields.","Add unit tests that verify format fields survive copy()."],"tags":["table-sink","format-factory","writer-creation","table-exception","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}