{"record":{"id":"d98582a5b2fc69a0","repo":"apache/flink","slug":"duplicated-serializer-for-the-same-class","errorCode":null,"errorMessage":"Duplicated serializer for the same class.","messagePattern":"Duplicated serializer for the same class\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java","lineNumber":398,"sourceCode":"        }\n    }\n\n    private void parseSerializationConfig(\n            ClassLoader classLoader, List<String> serializationConfigs) {\n        final LinkedHashMap<Class<?>, Map<String, String>> serializationConfigByClass =\n                serializationConfigs.stream()\n                        .map(ConfigurationUtils::parseStringToMap)\n                        .flatMap(m -> m.entrySet().stream())\n                        .collect(\n                                Collectors.toMap(\n                                        e ->\n                                                loadClass(\n                                                        e.getKey(),\n                                                        classLoader,\n                                                        \"Could not load class for serialization config\"),\n                                        e -> ConfigurationUtils.parseStringToMap(e.getValue()),\n                                        (v1, v2) -> {\n                                            throw new IllegalArgumentException(\n                                                    \"Duplicated serializer for the same class.\");\n                                        },\n                                        LinkedHashMap::new));\n        for (Map.Entry<Class<?>, Map<String, String>> entry :\n                serializationConfigByClass.entrySet()) {\n            Class<?> type = entry.getKey();\n            Map<String, String> config = entry.getValue();\n            String configType = config.get(\"type\");\n            if (configType == null) {\n                throw new IllegalArgumentException(\"Serializer type not specified for \" + type);\n            }\n            switch (configType) {\n                case \"pojo\":\n                    registerPojoType(type);\n                    break;\n                case \"kryo\":\n                    parseAndRegisterKryoType(classLoader, type, config);\n                    break;","sourceCodeStart":380,"sourceCodeEnd":416,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java#L380-L416","documentation":"Thrown inside the Collectors.toMap merge function during parseSerializationConfig when two or more entries in the serialization-config map target the same class. The merge function is only invoked on key collisions, so this means the configured class appears more than once with different serializer definitions. The system cannot decide which serializer wins, so it fails fast.","triggerScenarios":"Setting pipeline.serialization-config with the same fully-qualified class name appearing as a key in two separate config entries. For example, listing 'com.example.MyType' twice in the config string.","commonSituations":"Merging serialization configs from multiple sources (default + override) without deduplication; copy-paste errors in config files; YAML anchors producing duplicate keys after expansion.","solutions":["Deduplicate the serialization-config entries so each class appears exactly once.","If merging configs programmatically, use a Map to ensure last-write-wins or detect conflicts before passing to Flink.","Audit the full config string for repeated class names."],"exampleFix":"# before (duplicate class)\npipeline.serialization-config: \"class:com.example.MyType{type:kryo};class:com.example.MyType{type:pojo}\"\n\n# after (single entry)\npipeline.serialization-config: \"class:com.example.MyType{type:kryo}\"","handlingStrategy":"validation","validationCode":"// Deduplicate config entries before passing to Flink\nMap<String, Map<String, String>> deduped = new LinkedHashMap<>();\nfor (Map.Entry<String, Map<String, String>> e : rawEntries.entrySet()) {\n    if (deduped.containsKey(e.getKey())) {\n        throw new IllegalStateException(\"Duplicate serializer entry for: \" + e.getKey());\n    }\n    deduped.put(e.getKey(), e.getValue());\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Audit serialization-config strings for repeated class names.","When merging config sources, deduplicate by class name with explicit conflict handling."],"tags":["serialization","configuration","duplicate","parsing"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}