{"record":{"id":"751a7360a833f13c","repo":"apache/flink","slug":"unable-to-serialize-default-value-of-type","errorCode":null,"errorMessage":"Unable to serialize default value of type {}.","messagePattern":"Unable to serialize default value of type (.+?)\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java","lineNumber":411,"sourceCode":"            out.writeBoolean(false);\n        } else {\n            TypeSerializer<T> serializer = serializerAtomicReference.get();\n            checkNotNull(serializer, \"Serializer not initialized.\");\n\n            // we have a default value\n            out.writeBoolean(true);\n\n            byte[] serializedDefaultValue;\n            try (ByteArrayOutputStream baos = new ByteArrayOutputStream();\n                    DataOutputViewStreamWrapper outView = new DataOutputViewStreamWrapper(baos)) {\n\n                TypeSerializer<T> duplicateSerializer = serializer.duplicate();\n                duplicateSerializer.serialize(defaultValue, outView);\n\n                outView.flush();\n                serializedDefaultValue = baos.toByteArray();\n            } catch (Exception e) {\n                throw new IOException(\n                        \"Unable to serialize default value of type \"\n                                + defaultValue.getClass().getSimpleName()\n                                + \".\",\n                        e);\n            }\n\n            out.writeInt(serializedDefaultValue.length);\n            out.write(serializedDefaultValue);\n        }\n    }\n\n    private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {\n        // read the non-transient fields\n        in.defaultReadObject();\n\n        // read the default value field\n        boolean hasDefaultValue = in.readBoolean();\n        if (hasDefaultValue) {","sourceCodeStart":393,"sourceCodeEnd":429,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/state/StateDescriptor.java#L393-L429","documentation":"During Java serialization of the StateDescriptor (writeObject), Flink serializes the non-null defaultValue using a duplicate of the descriptor's TypeSerializer. If serializer.serialize(defaultValue, outView) throws (e.g., the value type does not match what the serializer expects, or the serializer implementation has a bug), the exception is wrapped in an IOException with the value's simple class name.","triggerScenarios":"The defaultValue was set with an instance whose runtime type differs from the serializer's T (e.g., assigning a subclass the serializer cannot handle); a custom TypeSerializer that fails on a particular value; serializing a descriptor whose serializer and default value were configured inconsistently.","commonSituations":"Distributing the job graph (Java serialization of descriptors to TaskManagers); checkpointing the descriptor; a default value set via a builder that bypasses type checks; upgrading a serializer snapshot while the in-memory default value is stale.","solutions":["Ensure defaultValue's runtime type is exactly the type the descriptor's serializer handles — do not pass a subclass instance as the default.","Test the serializer round-trip on the default value in isolation (serialize/deserialize) before constructing the descriptor.","If using a custom TypeSerializer, verify serialize/deserialize are symmetric and handle the default value.","Avoid setting a default value of a type that the serializer cannot represent; pass null instead and handle missing-state logic in the function."],"exampleFix":"// before\nnew ValueStateDescriptor<>(\"n\", MyPojo.class, new MyPojoSubclass()); // type mismatch\n\n// after\nnew ValueStateDescriptor<>(\"n\", MyPojo.class, new MyPojo());","handlingStrategy":"type-guard","validationCode":"// Verify the default value type matches the descriptor's type before assigning\nT defaultValue = ...;\nif (!descriptorTypeClass.isInstance(defaultValue)) {\n    throw new IllegalArgumentException(\"Default value type mismatch\");\n}\n// Round-trip test the serializer on the default value\nTypeSerializer<T> ser = descriptor.getTypeInformation().createSerializer(cfg);\ntry (ByteArrayOutputStream out = new ByteArrayOutputStream()) {\n    ser.serialize(defaultValue, new DataOutputViewStreamWrapper(out));\n    T copy = ser.deserialize(new DataInputViewStreamWrapper(\n        new ByteArrayInputStream(out.toByteArray())));\n}","typeGuard":"// Ensure defaultValue is exactly the serializer's type\nClass<?> t = descriptor.getTypeInformation().getTypeClass();\nif (!t.isInstance(defaultValue)) {\n    throw new ClassCastException(defaultValue + \" is not a \" + t);\n}","tryCatchPattern":"// Serialization happens during job graph distribution; catch in tests, not in prod.\ntry {\n    env.execute(); // triggers descriptor serialization\n} catch (Exception e) {\n    if (e.getCause() instanceof IOException\n        && e.getMessage().contains(\"Unable to serialize default value\")) {\n        // fix the default value type and rebuild\n    }\n}","preventionTips":["Never pass a subclass instance as a default value when the serializer targets the superclass.","Add a serialization round-trip test for every descriptor that carries a non-null default.","Keep the default value's type equal to the descriptor's declared type."],"tags":["state-descriptor","serialization","default-value","type-serializer"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}