{"record":{"id":"97b04584a88e1d81","repo":"apache/flink","slug":"concurrent-access-to-kryoserializer-thread-1-th","errorCode":null,"errorMessage":"Concurrent access to KryoSerializer. Thread 1: {threadName1} , Thread 2: {threadName2}","messagePattern":"Concurrent access to KryoSerializer\\. Thread 1: (.+?) , Thread 2: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java","lineNumber":706,"sourceCode":"                    \"Could not clone serializer instance of class \" + original.getClass(), ex);\n        }\n    }\n\n    // --------------------------------------------------------------------------------------------\n    // For testing\n    // --------------------------------------------------------------------------------------------\n\n    private void enterExclusiveThread() {\n        // we use simple get, check, set here, rather than CAS\n        // we don't need lock-style correctness, this is only a sanity-check and we thus\n        // favor speed at the cost of some false negatives in this check\n        Thread previous = currentThread;\n        Thread thisThread = Thread.currentThread();\n\n        if (previous == null) {\n            currentThread = thisThread;\n        } else if (previous != thisThread) {\n            throw new IllegalStateException(\n                    \"Concurrent access to KryoSerializer. Thread 1: \"\n                            + thisThread.getName()\n                            + \" , Thread 2: \"\n                            + previous.getName());\n        }\n    }\n\n    private void exitExclusiveThread() {\n        currentThread = null;\n    }\n\n    @VisibleForTesting\n    public Kryo getKryo() {\n        checkKryoInitialized();\n        return this.kryo;\n    }\n}\n","sourceCodeStart":688,"sourceCodeEnd":724,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java#L688-L724","documentation":"KryoSerializer is not thread-safe; it tracks the current accessing thread with a best-effort check (enterExclusiveThread/exitExclusiveThread) around serialize/deserialize. If serialize() or deserialize() is entered while a different thread is already recorded (previous != null && previous != thisThread), it throws IllegalStateException naming both threads. Note the check is deliberately non-atomic, so genuine races may instead produce corrupted Kryo state rather than this exception.","triggerScenarios":"Two threads calling serialize()/deserialize() on the SAME KryoSerializer instance concurrently; sharing one serializer instance across parallel subtasks, a thread pool, or an async snapshotting thread plus the mailbox thread; re-entrant serialization from a custom Kryo serializer that itself uses the same KryoSerializer.","commonSituations":"Caching a TypeSerializer in a static/instance field used by multiple worker threads in an application or in a source/sink that serializes from an I/O thread; reusing serializers across operator subtasks instead of duplicating them; using the same KryoSerializer from a Kafka callback thread and the main pipeline thread.","solutions":["Give each thread its own serializer: call serializer.duplicate() (TypeSerializer API) per thread instead of sharing one instance.","Remove static/shared caching of the TypeSerializer — obtain it per subtask/per thread from the TypeInformation.","Synchronize external access if duplication is impossible: wrap serialize/deserialize calls in a lock so only one thread enters at a time.","In custom Kryo serializers, never re-enter the outer serializer; use the Kryo instance passed to the callback."],"exampleFix":"// before\nprivate static final TypeSerializer<MyEvent> SER = TypeInfoFactory...createSerializer();\n// shared across pool threads -> IllegalStateException\n\n// after\nprivate final TypeSerializer<MyEvent> ser = baseSerializer.duplicate(); // one per thread/subtask","handlingStrategy":"validation","validationCode":"// Before sharing, give each thread its own copy — TypeSerializer.duplicate() exists for this.\nTypeSerializer<T> threadLocalSer = baseSer.duplicate();","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Treat TypeSerializer instances as thread-confined; never cache them in static fields.","Call duplicate() whenever a serializer crosses a thread boundary (pools, async sinks, snapshot threads).","Remember the check is best-effort: absence of this exception does not prove thread safety."],"tags":["thread-safety","kryo","concurrency","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}