{"record":{"id":"e0526cd19d0e2daf","repo":"apache/flink","slug":"undefined-compatibility-type","errorCode":null,"errorMessage":"Undefined compatibility type.","messagePattern":"Undefined compatibility type\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java","lineNumber":121,"sourceCode":"            TypeSerializerSchemaCompatibility<?> compatibility =\n                    resolveCompatibility(\n                            newNestedSerializerSnapshots[i], oldNestedSerializerSnapshots[i]);\n\n            // if any one of the new nested serializers is incompatible, we can just short circuit\n            // the result\n            if (compatibility.isIncompatible()) {\n                return IntermediateCompatibilityResult.definedIncompatibleResult();\n            }\n\n            if (compatibility.isCompatibleAfterMigration()) {\n                nestedSerializerRequiresMigration = true;\n            } else if (compatibility.isCompatibleWithReconfiguredSerializer()) {\n                hasReconfiguredNestedSerializers = true;\n                nestedSerializers[i] = compatibility.getReconfiguredSerializer();\n            } else if (compatibility.isCompatibleAsIs()) {\n                nestedSerializers[i] = newNestedSerializerSnapshots[i].restoreSerializer();\n            } else {\n                throw new IllegalStateException(\"Undefined compatibility type.\");\n            }\n        }\n\n        if (nestedSerializerRequiresMigration) {\n            return IntermediateCompatibilityResult.definedCompatibleAfterMigrationResult();\n        }\n\n        if (hasReconfiguredNestedSerializers) {\n            return IntermediateCompatibilityResult.undefinedReconfigureResult(nestedSerializers);\n        }\n\n        // ends up here if everything is compatible as is\n        return IntermediateCompatibilityResult.definedCompatibleAsIsResult(nestedSerializers);\n    }\n\n    public static class IntermediateCompatibilityResult<T> {\n\n        private final TypeSerializerSchemaCompatibility.Type compatibilityType;","sourceCodeStart":103,"sourceCodeEnd":139,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java#L103-L139","documentation":"Thrown by CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult when a nested serializer's resolveCompatibility call returns a TypeSerializerSchemaCompatibility that matches none of the four recognized outcomes (incompatible, compatibleAfterMigration, compatibleWithReconfiguredSerializer, compatibleAsIs). This is an internal invariant violation indicating a custom TypeSerializerSnapshot returned an invalid or null compatibility result.","triggerScenarios":"A custom TypeSerializerSnapshot implementation whose resolveSchemaCompatibility method returns null or a TypeSerializerSchemaCompatibility built through an unsupported/undocumented code path. This should not happen with the standard factory methods (compatibleAsIs, compatibleAfterMigration, incompatible, compatibleWithReconfiguredSerializer).","commonSituations":"Implementing a custom serializer and its snapshot incorrectly — returning a raw or improperly-constructed TypeSerializerSchemaCompatibility. A library/framework wrapping Flink serializers with a snapshot that returns an unexpected enum variant. Regression after upgrading a custom serializer library.","solutions":["Audit any custom TypeSerializerSnapshot.resolveSchemaCompatibility implementation to ensure it returns only values from the four standard factory methods.","If using a third-party serializer library, upgrade it or check its Flink integration for a known bug.","Replace the custom snapshot with one that delegates to SimpleTypeSerializerSnapshot or CompositeTypeSerializerSnapshot to avoid hand-rolling compatibility logic.","Add a unit test that calls resolveSchemaCompatibility with old and new snapshots and asserts the returned Type."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Unit-test your custom snapshot to ensure it returns a standard compatibility result\nTypeSerializerSchemaCompatibility<?> c =\n    newSnapshot.resolveSchemaCompatibility(oldSnapshot);\nif (!c.isCompatibleAsIs() && !c.isCompatibleAfterMigration()\n        && !c.isCompatibleWithReconfiguredSerializer() && !c.isIncompatible()) {\n    throw new AssertionError(\"Custom snapshot returned an invalid compatibility result\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    backend.restore(state);\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"Undefined compatibility type\")) {\n        // a custom TypeSerializerSnapshot returned an invalid result\n        log.error(\"A custom serializer snapshot returned an unsupported compatibility result\", e);\n    }\n    throw e;\n}","preventionTips":["When implementing a custom TypeSerializerSnapshot, only return results from the four standard factory methods.","Prefer extending SimpleTypeSerializerSnapshot or CompositeTypeSerializerSnapshot over hand-rolling a snapshot.","Add unit tests for resolveSchemaCompatibility covering old and new snapshot versions."],"tags":["serialization","state","schema-compatibility","internal-invariant","custom-serializer"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}