{"record":{"id":"6a1cb830196150e0","repo":"apache/cassandra","slug":"stream-receive-task-s-of-cf-s-already-finished","errorCode":null,"errorMessage":"Stream receive task %s of cf %s already finished.","messagePattern":"Stream receive task (.+?) of cf (.+?) already finished\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"src/java/org/apache/cassandra/streaming/StreamReceiveTask.java","lineNumber":121,"sourceCode":"            done = true;\n            executor.submit(new OnCompletionRunnable(this));\n        }\n    }\n\n    public int getTotalNumberOfFiles()\n    {\n        return totalStreams;\n    }\n\n    public long getTotalSize()\n    {\n        return totalSize;\n    }\n\n    public synchronized StreamReceiver getReceiver()\n    {\n        if (done)\n            throw new RuntimeException(String.format(\"Stream receive task %s of cf %s already finished.\", session.planId(), tableId));\n        return receiver;\n    }\n\n    private static class OnCompletionRunnable implements Runnable\n    {\n        private final StreamReceiveTask task;\n\n        public OnCompletionRunnable(StreamReceiveTask task)\n        {\n            this.task = task;\n        }\n\n        public void run()\n        {\n            try\n            {\n                if (ColumnFamilyStore.getIfExists(task.tableId) == null)\n                {","sourceCodeStart":103,"sourceCodeEnd":139,"githubUrl":"https://github.com/apache/cassandra/blob/88fd0f6a0eaed8943f05ac9e8f947882b8ddc8f1/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java#L103-L139","documentation":"StreamReceiveTask wraps the receiving side of one streaming session for one table. Its getReceiver() hands out the StreamReceiver that incoming file messages are written through. Once the task is done (all files received and the completion runnable ran), the receiver is invalidated, and any late-arriving lookup throws this RuntimeException to prevent use-after-completion of the receiver.","triggerScenarios":"Calling StreamReceiveTask.getReceiver() after task.done has been set, i.e. after all expected incoming streams arrived and OnCompletionRunnable finished the task; typically a duplicate or late IncomingStreamMessage arriving for an already-completed stream task.","commonSituations":"Network retries delivering a file message twice; sender and receiver disagree on the number of files so an extra message lands after completion; mixing node versions during upgrades where streaming protocol edge cases resend messages.","solutions":["Check StreamReceiveTask task state (done) before calling getReceiver(), or make getReceiver() calls idempotent in the receive path.","Verify the sender's file/section count matches the receiver's StreamSummary so no extra messages are sent after completion.","Retry the streaming session (StreamSession failure handling will cancel it) rather than reusing the finished task.","If reproducible across an upgrade, confirm both nodes run compatible Cassandra streaming versions."],"exampleFix":"// before\nStreamReceiver receiver = task.getReceiver(); // may throw if already finished\n// after\nif (!task.isDone()) {\n    StreamReceiver receiver = task.getReceiver();\n    receiver.received(stream);\n} else {\n    logger.warn(\"Ignoring late stream message for finished task\");\n}","handlingStrategy":"try-catch","validationCode":"if (task == null || task.isDone()) { skipOrLog(); }","typeGuard":null,"tryCatchPattern":"try { StreamReceiver r = task.getReceiver(); r.received(stream); } catch (RuntimeException e) { if (e.getMessage().contains(\"already finished\")) { logger.warn(\"Late stream message for finished task\"); } else { throw e; } }","preventionTips":["Ensure receiver vs sender file counts match so no extra messages arrive after completion.","Make completion handlers idempotent and tolerant of duplicate/late messages.","Keep cluster node versions aligned during upgrades."],"tags":["streaming","lifecycle","use-after-completion"],"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"}