{"record":{"id":"34bd13e1b26a6a74","repo":"apache/flink","slug":"failed-to-create-enumerator-for-sourceindex-curre","errorCode":null,"errorMessage":"Failed to create enumerator for sourceIndex={currentSourceIndex}","messagePattern":"Failed to create enumerator for sourceIndex=(.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java","lineNumber":306,"sourceCode":"                        currentSourceIndex,\n                        context,\n                        readerSourceIndex,\n                        switchedSources,\n                        sources.size());\n        try {\n            if (restoredEnumeratorState == null) {\n                currentEnumerator = source.createEnumerator(delegatingContext);\n            } else {\n                LOG.info(\"Restoring enumerator for sourceIndex={}\", currentSourceIndex);\n                Object nestedEnumState =\n                        currentEnumeratorCheckpointSerializer.deserialize(\n                                restoredEnumeratorState.getWrappedStateSerializerVersion(),\n                                restoredEnumeratorState.getWrappedState());\n                currentEnumerator = source.restoreEnumerator(delegatingContext, nestedEnumState);\n                restoredEnumeratorState = null;\n            }\n        } catch (Exception e) {\n            throw new RuntimeException(\n                    \"Failed to create enumerator for sourceIndex=\" + currentSourceIndex, e);\n        }\n        LOG.info(\"Starting enumerator for sourceIndex={}\", currentSourceIndex);\n        context.setIsProcessingBacklog(currentSourceIndex < sources.size() - 1);\n        currentEnumerator.start();\n    }\n\n    /**\n     * The {@link SplitEnumeratorContext} that is provided to the currently active enumerator.\n     *\n     * <p>This context is used to wrap the splits into {@link HybridSourceSplit} and track\n     * assignment to readers.\n     */\n    private static class SplitEnumeratorContextProxy<SplitT extends SourceSplit>\n            implements SplitEnumeratorContext<SplitT> {\n        private static final Logger LOG =\n                LoggerFactory.getLogger(SplitEnumeratorContextProxy.class);\n","sourceCodeStart":288,"sourceCodeEnd":324,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java#L288-L324","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before: factory throws because context previous state is wrong type\nsource = factory.create(switchContext);\n// after: guard the switch state before building the next source\nHybridSource.SourceSwitchContext<?> ctx = ...;\nif (ctx.getPreviousEnumerator() instanceof ExpectedState) {\n    source = factory.create(ctx);\n} else {\n    throw new IllegalStateException(\"Unexpected previous enumerator state for sourceIndex=\" + index);\n}","handlingStrategy":"try-catch","validationCode":"// Before switching sources, sanity-check the factory and prior enumerator state\nHybridSource.SourceSwitchContext<?> ctx = switchContext;\nObject prev = ctx.getPreviousEnumerator();\nif (prev != null && !expectedStateClass.isInstance(prev)) {\n    throw new IllegalStateException(\n        \"Previous enumerator state for sourceIndex=\" + index\n        + \" is \" + prev.getClass() + \" but \" + expectedStateClass + \" was expected\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    hybridSource.createEnumerator(context);\n} catch (RuntimeException e) {\n    Throwable root = e.getCause() != null ? e.getCause() : e;\n    if (root instanceof org.apache.flink.util.FlinkRuntimeException) {\n        // sub-source failed; report index from message and fail the job\n        throw e;\n    }\n    throw e;\n}","preventionTips":["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."],"tags":["hybrid-source","checkpoint-restore","source-enumerator","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}