apache/flink · error · UnsupportedOperationException

Unsupported input message: {value}

Error message

Unsupported input message: {value}

What it means

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.

Source

Thrown at flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/compact/CompactCoordinator.java:149

            InputFile file = (InputFile) value;
            currentInputFiles
                    .computeIfAbsent(file.getPartition(), k -> new ArrayList<>())
                    .add(file.getFile());
        } else if (value instanceof EndCheckpoint) {
            EndCheckpoint endCheckpoint = (EndCheckpoint) value;
            if (inputTaskTracker == null) {
                inputTaskTracker = new TaskTracker(endCheckpoint.getNumberOfTasks());
            }

            // ensure all files are ready to be compacted.
            boolean triggerCommit =
                    inputTaskTracker.add(
                            endCheckpoint.getCheckpointId(), endCheckpoint.getTaskId());
            if (triggerCommit) {
                commitUpToCheckpoint(endCheckpoint.getCheckpointId());
            }
        } else {
            throw new UnsupportedOperationException("Unsupported input message: " + value);
        }
    }

    private void commitUpToCheckpoint(long checkpointId) {
        Map<Long, Map<String, List<Path>>> headMap = inputFiles.headMap(checkpointId, true);
        for (Map.Entry<Long, Map<String, List<Path>>> entry : headMap.entrySet()) {
            coordinate(entry.getKey(), entry.getValue());
        }
        if (checkpointId == Long.MAX_VALUE) {
            coordinate(checkpointId, currentInputFiles);
            currentInputFiles.clear();
        }
        headMap.clear();
    }

    /** Do stable compaction coordination. */
    private void coordinate(long checkpointId, Map<String, List<Path>> partFiles) {
        Function<Path, Long> sizeFunc =

View on GitHub (pinned to 2f3c205e92)

Solutions

  1. Ensure only InputFile and EndCheckpoint messages reach the CompactCoordinator — do not inject custom messages into the compaction operator's input stream.
  2. If this appears after a Flink upgrade, take a fresh savepoint and restart without restoring the old one.
  3. Verify there are no classpath conflicts on org.apache.flink.connector.file.table.stream.CoordinatorInput subtypes.
Defensive patterns

Strategy: try-catch

Try / catch

// This is an internal invariant violation; catch to log and fail fast
try {
    coordinator.processElement(record);
} catch (UnsupportedOperationException e) {
    if (e.getMessage().startsWith("Unsupported input message")) {
        LOG.error("CompactCoordinator received unexpected message type: {}", record.getValue());
        throw e; // re-throw — this indicates a version or class-loading bug
    }
    throw e;
}

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/232498b108816426. Report an issue: GitHub.