{"record":{"id":"232498b108816426","repo":"apache/flink","slug":"unsupported-input-message-value-232498","errorCode":null,"errorMessage":"Unsupported input message: {value}","messagePattern":"Unsupported input message: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/compact/CompactCoordinator.java","lineNumber":149,"sourceCode":"            InputFile file = (InputFile) value;\n            currentInputFiles\n                    .computeIfAbsent(file.getPartition(), k -> new ArrayList<>())\n                    .add(file.getFile());\n        } else if (value instanceof EndCheckpoint) {\n            EndCheckpoint endCheckpoint = (EndCheckpoint) value;\n            if (inputTaskTracker == null) {\n                inputTaskTracker = new TaskTracker(endCheckpoint.getNumberOfTasks());\n            }\n\n            // ensure all files are ready to be compacted.\n            boolean triggerCommit =\n                    inputTaskTracker.add(\n                            endCheckpoint.getCheckpointId(), endCheckpoint.getTaskId());\n            if (triggerCommit) {\n                commitUpToCheckpoint(endCheckpoint.getCheckpointId());\n            }\n        } else {\n            throw new UnsupportedOperationException(\"Unsupported input message: \" + value);\n        }\n    }\n\n    private void commitUpToCheckpoint(long checkpointId) {\n        Map<Long, Map<String, List<Path>>> headMap = inputFiles.headMap(checkpointId, true);\n        for (Map.Entry<Long, Map<String, List<Path>>> entry : headMap.entrySet()) {\n            coordinate(entry.getKey(), entry.getValue());\n        }\n        if (checkpointId == Long.MAX_VALUE) {\n            coordinate(checkpointId, currentInputFiles);\n            currentInputFiles.clear();\n        }\n        headMap.clear();\n    }\n\n    /** Do stable compaction coordination. */\n    private void coordinate(long checkpointId, Map<String, List<Path>> partFiles) {\n        Function<Path, Long> sizeFunc =","sourceCodeStart":131,"sourceCodeEnd":167,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/compact/CompactCoordinator.java#L131-L167","documentation":"Thrown by CompactCoordinator.processElement when the incoming StreamRecord value is neither an InputFile nor an EndCheckpoint. The coordinator's input type is the sealed hierarchy CoordinatorInput with exactly those two subtypes; any other value indicates a bug in the upstream operator, a class-loading mismatch, or an incompatible serialized message reaching the coordinator.","triggerScenarios":"A custom operator upstream of the compaction coordinator emits a non-CoordinatorInput message; or a Flink version upgrade changes the CoordinatorInput class structure causing deserialization into an unhandled subtype.","commonSituations":"Running a job restored from a savepoint taken with an incompatible Flink version; a classpath conflict where CoordinatorInput subtypes resolve to different class versions across the coordinator and subtask classloaders.","solutions":["Ensure only InputFile and EndCheckpoint messages reach the CompactCoordinator — do not inject custom messages into the compaction operator's input stream.","If this appears after a Flink upgrade, take a fresh savepoint and restart without restoring the old one.","Verify there are no classpath conflicts on org.apache.flink.connector.file.table.stream.CoordinatorInput subtypes."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// This is an internal invariant violation; catch to log and fail fast\ntry {\n    coordinator.processElement(record);\n} catch (UnsupportedOperationException e) {\n    if (e.getMessage().startsWith(\"Unsupported input message\")) {\n        LOG.error(\"CompactCoordinator received unexpected message type: {}\", record.getValue());\n        throw e; // re-throw — this indicates a version or class-loading bug\n    }\n    throw e;\n}","preventionTips":["Do not inject custom messages into the compaction operator's input stream.","After Flink upgrades, take a fresh savepoint rather than restoring across versions.","Verify no classpath conflicts on CoordinatorInput subtypes across the cluster."],"tags":["filesystem-connector","compaction","internal-api","class-loading"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}