{"record":{"id":"8a7ff539169a7bfa","repo":"apache/flink","slug":"failed-to-deserialize-csv-row-s","errorCode":null,"errorMessage":"Failed to deserialize CSV row '%s'.","messagePattern":"Failed to deserialize CSV row '(.+?)'\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDataDeserializationSchema.java","lineNumber":257,"sourceCode":"                    disabledFeatures.isEmpty()\n                            ? EnumSet.noneOf(CsvParser.Feature.class)\n                            : EnumSet.copyOf(disabledFeatures));\n        }\n    }\n\n    @Override\n    public RowData deserialize(@Nullable byte[] message) throws IOException {\n        if (message == null) {\n            return null;\n        }\n        try {\n            final JsonNode root = objectReader.readValue(message);\n            return (RowData) runtimeConverter.convert(root);\n        } catch (Throwable t) {\n            if (ignoreParseErrors) {\n                return null;\n            }\n            throw new IOException(\n                    String.format(\"Failed to deserialize CSV row '%s'.\", new String(message)), t);\n        }\n    }\n\n    @Override\n    public boolean isEndOfStream(RowData nextElement) {\n        return false;\n    }\n\n    @Override\n    public TypeInformation<RowData> getProducedType() {\n        return resultTypeInfo;\n    }\n\n    @Override\n    public boolean equals(Object o) {\n        if (this == o) {\n            return true;","sourceCodeStart":239,"sourceCodeEnd":275,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDataDeserializationSchema.java#L239-L275","documentation":"Thrown by CsvRowDataDeserializationSchema.deserialize() when parsing the Kafka byte[] into a JsonNode or running the runtime converter fails and ignoreParseErrors is false. The raw CSV line is embedded in the message and the underlying Jackson/converter exception is chained. This is the classic 'dirty CSV row vs declared schema' failure for the csv format.","triggerScenarios":"A CSV line with wrong column count, non-numeric text in an INT column, malformed quoting, or a timestamp/date string that doesn't match the expected format; a schema change upstream (new column) while the Flink table schema stayed old; empty lines on the topic.","commonSituations":"Production topics fed by heterogeneous producers; late-arriving schema evolution; delimiter/quote misconfiguration (e.g. actual delimiter ';' but option default ','); locale-specific number formats.","solutions":["If occasional dirty rows are acceptable, set 'csv.ignore-parse-errors'='true' (rows are skipped/logged) — most common fix.","Otherwise correct the row or the producer so values match the declared column types.","Verify csv.field-delimiter / quote / escape options match the actual file format.","Reproduce offline: the offending CSV text is printed in the exception; feed it to a local CsvReader with the same schema."],"exampleFix":"-- before\n'format'='csv'\n\n-- after\n'format'='csv',\n'csv.ignore-parse-errors'='true'","handlingStrategy":"fallback","validationCode":"-- Opt-in tolerance when dirty rows are acceptable:\n-- 'csv.ignore-parse-errors' = 'true'\n-- For strict pipelines, sample and validate data first:\n-- SELECT COUNT(*) FROM kafka_t /*+ OPTIONS('scan.startup.mode'='earliest') */ WHERE <type-check predicates>;","typeGuard":null,"tryCatchPattern":"catch (IOException e) { if (ignoreParseErrors) { log.warn(\"skipping bad CSV row: {}\", e.getMessage()); return null; } throw e; } — or simply enable the format's ignore-parse-errors option","preventionTips":["Set ignore-parse-errors for best-effort pipelines; stay strict and alert for financial ones","Contract test producer output against the table schema in CI","Verify delimiter/quote options against real samples with kcat"],"tags":["csv","deserialization","dirty-data","kafka","parsing"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}