apache/cassandra · error · RuntimeException

stream has been closed, cannot send

Error message

stream has been closed, cannot send %s

What it means

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.

Solutions

1. Check StreamingMultiplexedChannel closed state (or sendControlMessage return) before sending, and tolerate the session being already terminated.
2. Fix/serialize the session teardown path so messages aren't sent after close() — check for races between complete()/closeSession and message sends.
3. Investigate why the channel closed early (peer logs, prior exceptions on the same planId) — often a network drop or peer-side failure.
4. Re-run the streaming operation; a closed channel cannot be revived.

Example fix

// before
channel.sendMessage(streamingChannel, message); // may throw if closed
// after
if (!channel.isClosed()) {
    channel.sendMessage(streamingChannel, message);
} else {
    session.closeSession(State.Type.FAILED); // session already torn down
}

if (channel.isClosed()) { /* do not send; fail or recreate the session */ }

try { channel.sendMessage(streamingChannel, msg); } catch (RuntimeException e) { if (e.getMessage().startsWith("stream has been closed")) { session.closeSession(State.Type.FAILED); } else { throw e; } }

When it happens

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

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

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• A node required to move the data consistently is down
• Can not start range streaming as all candidates
• Can't join the ring because bootstrap hasn't completed.
• Cannot create materialized view
• Cannot send stream data messages for preview streaming…

AI-assisted analysis of apache/cassandra@88fd0f6a0e (2026-09-10). Data as JSON: /api/errors/2b4805afaf74e9d2. Report an issue: GitHub.

Appendix: source

Thrown at src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java:214

    public Future<?> sendControlMessage(StreamMessage message)
    {
        try
        {
            setupControlMessageChannel();
            return sendMessage(controlChannel, message);
        }
        catch (Exception e)
        {
            close();
            session.onError(e);
            return ImmediateFuture.failure(e);
        }

    }
    public Future<?> sendMessage(StreamingChannel channel, StreamMessage message)
    {
        if (closed)
            throw new RuntimeException("stream has been closed, cannot send " + message);

        if (message instanceof OutgoingStreamMessage)
        {
            if (session.isPreview())
                throw new RuntimeException("Cannot send stream data messages for preview streaming sessions");
            if (logger.isDebugEnabled())
                logger.debug("{} Sending {}", createLogTag(session), message);

            InetAddressAndPort connectTo = factory.supportsPreferredIp() ? SystemKeyspace.getPreferredIP(to) : to;
            return fileTransferExecutor.submit(new FileStreamTask((OutgoingStreamMessage) message, connectTo));
        }

        try
        {
            Future<?> promise = channel.send(outSupplier -> {
                // we anticipate that the control messages are rather small, so allocating a ByteBuf shouldn't  blow out of memory.
                long messageSize = serializedSize(message, messagingVersion);
                if (messageSize > 1 << 30)

View on GitHub (pinned to 88fd0f6a0e)