{"record":{"id":"97492b740a270d1c","repo":"apache/cassandra","slug":"stream-s-is-finished-with-state-s","errorCode":null,"errorMessage":"Stream %s is finished with state %s","messagePattern":"Stream (.+?) is finished with state (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"src/java/org/apache/cassandra/streaming/StreamSession.java","lineNumber":481,"sourceCode":"        //Do we need to unwrap here also or is that just making it worse?\n        //Range and if it's transient\n        RangesAtEndpoint unwrappedRanges = replicas.unwrap();\n        List<OutgoingStream> streams = getOutgoingStreamsForRanges(unwrappedRanges, stores, pendingRepair, previewKind);\n\n        addTransferStreams(streams);\n        Set<Range<Token>> toBeUpdated = transferredRangesPerKeyspace.get(keyspace);\n        if (toBeUpdated == null)\n        {\n            toBeUpdated = new HashSet<>();\n        }\n        toBeUpdated.addAll(replicas.ranges());\n        transferredRangesPerKeyspace.put(keyspace, toBeUpdated);\n    }\n\n    private void failIfFinished()\n    {\n        if (state().isFinalState())\n            throw new RuntimeException(String.format(\"Stream %s is finished with state %s\", planId(), state().name()));\n    }\n\n    private Collection<ColumnFamilyStore> getColumnFamilyStores(String keyspace, Collection<String> columnFamilies)\n    {\n        Collection<ColumnFamilyStore> stores = new HashSet<>();\n        // if columnfamilies are not specified, we add all cf under the keyspace\n        if (columnFamilies.isEmpty())\n        {\n            stores.addAll(Keyspace.open(keyspace).getColumnFamilyStores());\n        }\n        else\n        {\n            for (String cf : columnFamilies)\n                stores.add(Keyspace.open(keyspace).getColumnFamilyStore(cf));\n        }\n        return stores;\n    }\n","sourceCodeStart":463,"sourceCodeEnd":499,"githubUrl":"https://github.com/apache/cassandra/blob/88fd0f6a0eaed8943f05ac9e8f947882b8ddc8f1/src/java/org/apache/cassandra/streaming/StreamSession.java#L463-L499","documentation":"StreamSession guards every mutation (attach, transfer, message handling) with failIfFinished(), which throws if the session has already reached a final state (COMPLETED/FAILED). This prevents operating on a session whose streaming plan is over, e.g. messages or ranges arriving after completion.","triggerScenarios":"Calling attachInbound/attachOutbound, addTransferRanges, addTransferStreams, messageReceived, or prepareReceiving on a StreamSession whose state is in a final state — typically a control or data message arriving after the session already completed or failed.","commonSituations":"Slow/late messages delivered after session success; a session that failed earlier (node down, timeout) receiving follow-up traffic; coordinator retrying a stream request against a session that already ran.","solutions":["Check session state via session.state().isFinalState() before attaching ranges/streams or sending messages.","Create a new StreamSession (new planId) for retries instead of reusing a finished session.","Inspect earlier logs for the session's planId to see why it reached COMPLETED/FAILED before this call.","Fix network delays/duplicates or messaging retry settings that deliver messages after session completion."],"exampleFix":"// before\nsession.messageReceived(msg); // throws if finished\n// after\nif (!session.state().isFinalState()) {\n    session.messageReceived(msg);\n} else {\n    logger.warn(\"Dropping message for finished stream session {}\", session.planId());\n}","handlingStrategy":"validation","validationCode":"if (session.state().isFinalState()) { /* drop or start new session */ }","typeGuard":null,"tryCatchPattern":"try { session.messageReceived(msg); } catch (RuntimeException e) { if (e.getMessage().startsWith(\"Stream \") && e.getMessage().contains(\"is finished\")) { logger.warn(\"Message after stream end, ignoring\"); } else { throw e; } }","preventionTips":["Never reuse a StreamSession after it reaches a final state; create a fresh session for retries.","Log planId in all streaming operations to trace session lifecycle.","Suppress duplicate message delivery at the transport layer where possible."],"tags":["streaming","session-lifecycle","state-machine"],"backgroundTag":"invalid-state-transition","analyzedSha":"88fd0f6a0eaed8943f05ac9e8f947882b8ddc8f1","analyzedAt":"2026-09-10T07:29:22.284Z","contentChangedAt":"2026-09-10T07:29:22.284Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}