{"record":{"id":"2b4805afaf74e9d2","repo":"apache/cassandra","slug":"stream-has-been-closed-cannot-send-s","errorCode":null,"errorMessage":"stream has been closed, cannot send %s","messagePattern":"stream has been closed, cannot send (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java","lineNumber":214,"sourceCode":"    public Future<?> sendControlMessage(StreamMessage message)\n    {\n        try\n        {\n            setupControlMessageChannel();\n            return sendMessage(controlChannel, message);\n        }\n        catch (Exception e)\n        {\n            close();\n            session.onError(e);\n            return ImmediateFuture.failure(e);\n        }\n\n    }\n    public Future<?> sendMessage(StreamingChannel channel, StreamMessage message)\n    {\n        if (closed)\n            throw new RuntimeException(\"stream has been closed, cannot send \" + message);\n\n        if (message instanceof OutgoingStreamMessage)\n        {\n            if (session.isPreview())\n                throw new RuntimeException(\"Cannot send stream data messages for preview streaming sessions\");\n            if (logger.isDebugEnabled())\n                logger.debug(\"{} Sending {}\", createLogTag(session), message);\n\n            InetAddressAndPort connectTo = factory.supportsPreferredIp() ? SystemKeyspace.getPreferredIP(to) : to;\n            return fileTransferExecutor.submit(new FileStreamTask((OutgoingStreamMessage) message, connectTo));\n        }\n\n        try\n        {\n            Future<?> promise = channel.send(outSupplier -> {\n                // we anticipate that the control messages are rather small, so allocating a ByteBuf shouldn't  blow out of memory.\n                long messageSize = serializedSize(message, messagingVersion);\n                if (messageSize > 1 << 30)","sourceCodeStart":196,"sourceCodeEnd":232,"githubUrl":"https://github.com/apache/cassandra/blob/88fd0f6a0eaed8943f05ac9e8f947882b8ddc8f1/src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java#L196-L232","documentation":"StreamingMultiplexedChannel manages a session's control/data channels. sendMessage() throws this RuntimeException if the channel is already closed, preventing any message (control or data) from being sent over a terminated stream connection.","triggerScenarios":"Calling sendMessage() (directly or via sendControlMessage) after the channel's close() ran — e.g. a session thread sending a control message while another thread (or completion handler) concurrently closed the multiplexed channel.","commonSituations":"Race between session completion/failure and in-flight message sends; peer disconnected causing channel close while sends queued; a session failing (e.g. after a connect error) with subsequent messages attempted.","solutions":["Check StreamingMultiplexedChannel closed state (or sendControlMessage return) before sending, and tolerate the session being already terminated.","Fix/serialize the session teardown path so messages aren't sent after close() — check for races between complete()/closeSession and message sends.","Investigate why the channel closed early (peer logs, prior exceptions on the same planId) — often a network drop or peer-side failure.","Re-run the streaming operation; a closed channel cannot be revived."],"exampleFix":"// before\nchannel.sendMessage(streamingChannel, message); // may throw if closed\n// after\nif (!channel.isClosed()) {\n    channel.sendMessage(streamingChannel, message);\n} else {\n    session.closeSession(State.Type.FAILED); // session already torn down\n}","handlingStrategy":"try-catch","validationCode":"if (channel.isClosed()) { /* do not send; fail or recreate the session */ }","typeGuard":null,"tryCatchPattern":"try { channel.sendMessage(streamingChannel, msg); } catch (RuntimeException e) { if (e.getMessage().startsWith(\"stream has been closed\")) { session.closeSession(State.Type.FAILED); } else { throw e; } }","preventionTips":["Serialize teardown: stop sending before closing the channel.","Treat a closed channel as terminal; recreate the session instead of reusing it.","Correlate planId logs to find the earlier failure that closed the channel."],"tags":["streaming","channel-lifecycle","race-condition"],"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"}