apache/flink · error · RuntimeException

Failed to create enumerator for sourceIndex={currentSourceIn

Error message

Failed to create enumerator for sourceIndex={currentSourceIndex}

What it means

Thrown by HybridSource when it fails to either create a fresh SplitEnumerator or restore one from checkpoint state for the sub-source at the given sourceIndex. The try block wraps three operations: createEnumerator, deserialization of the nested enumerator checkpoint, and restoreEnumerator. Any exception from the underlying concrete Source (e.g. Kafka, FileSource) implementation is caught broadly and re-wrapped in a RuntimeException with the failing index.

Source

Thrown at flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java:306

                        currentSourceIndex,
                        context,
                        readerSourceIndex,
                        switchedSources,
                        sources.size());
        try {
            if (restoredEnumeratorState == null) {
                currentEnumerator = source.createEnumerator(delegatingContext);
            } else {
                LOG.info("Restoring enumerator for sourceIndex={}", currentSourceIndex);
                Object nestedEnumState =
                        currentEnumeratorCheckpointSerializer.deserialize(
                                restoredEnumeratorState.getWrappedStateSerializerVersion(),
                                restoredEnumeratorState.getWrappedState());
                currentEnumerator = source.restoreEnumerator(delegatingContext, nestedEnumState);
                restoredEnumeratorState = null;
            }
        } catch (Exception e) {
            throw new RuntimeException(
                    "Failed to create enumerator for sourceIndex=" + currentSourceIndex, e);
        }
        LOG.info("Starting enumerator for sourceIndex={}", currentSourceIndex);
        context.setIsProcessingBacklog(currentSourceIndex < sources.size() - 1);
        currentEnumerator.start();
    }

    /**
     * The {@link SplitEnumeratorContext} that is provided to the currently active enumerator.
     *
     * <p>This context is used to wrap the splits into {@link HybridSourceSplit} and track
     * assignment to readers.
     */
    private static class SplitEnumeratorContextProxy<SplitT extends SourceSplit>
            implements SplitEnumeratorContext<SplitT> {
        private static final Logger LOG =
                LoggerFactory.getLogger(SplitEnumeratorContextProxy.class);

View on GitHub (pinned to 2f3c205e92)

Solutions

Inspect the wrapped cause (Throwable#getCause) to find which of createEnumerator / restoreEnumerator / deserialize failed and the real root exception from the concrete connector.
If the cause is a checkpoint-state incompatibility, discard the savepoint/checkpoint or migrate via a version-compatible connector path, because the nested enumerator state cannot be deserialized.
Verify each HybridSource.SourceFactory#create builds a fully-configured concrete Source (correct auth, paths, offsets) for the index reported in the message.
If the failure is transient (e.g. broker/filesystem unreachable), fix the environment and restart the job; the enumerator creation is retried on recovery.
Ensure any custom Source passed into HybridSource implements restoreEnumerator consistent with its checkpoint serializer version.

Example fix

// before: factory throws because context previous state is wrong type
source = factory.create(switchContext);
// after: guard the switch state before building the next source
HybridSource.SourceSwitchContext<?> ctx = ...;
if (ctx.getPreviousEnumerator() instanceof ExpectedState) {
    source = factory.create(ctx);
} else {
    throw new IllegalStateException("Unexpected previous enumerator state for sourceIndex=" + index);
}

Defensive patterns

Strategy: try-catch

Validate before calling

// Before switching sources, sanity-check the factory and prior enumerator state
HybridSource.SourceSwitchContext<?> ctx = switchContext;
Object prev = ctx.getPreviousEnumerator();
if (prev != null && !expectedStateClass.isInstance(prev)) {
    throw new IllegalStateException(
        "Previous enumerator state for sourceIndex=" + index
        + " is " + prev.getClass() + " but " + expectedStateClass + " was expected");
}

Try / catch

try {
    hybridSource.createEnumerator(context);
} catch (RuntimeException e) {
    Throwable root = e.getCause() != null ? e.getCause() : e;
    if (root instanceof org.apache.flink.util.FlinkRuntimeException) {
        // sub-source failed; report index from message and fail the job
        throw e;
    }
    throw e;
}

Prevention

Pin consistent connector versions across the checkpoint and the restored job.
Unit-test each SourceFactory with a representative SourceSwitchContext before wiring it into HybridSource.
Log the sourceIndex from the message to narrow which sub-source failed.

When it happens

Trigger: Calling startSwitchedEnumerator / switchToSource during job initialization or failover recovery, where sources.get(currentSourceIndex).factory.create(...) yields a Source whose createEnumerator or restoreEnumerator throws. Also triggered when currentEnumeratorCheckpointSerializer.deserialize fails on a corrupt or version-incompatible wrapped enumerator state, or when the SourceSwitchContext.getPreviousEnumerator returns state the next source rejects.

Common situations: Checkpoint/savepoint restore after upgrading a connector whose enumerator state schema changed; a sub-source factory that misconfigures the delegate context; transient errors initializing the sub-source (auth, missing files, broker down); HybridSource chain where one source's restore path is not implemented.

Related errors

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