{"record":{"id":"2d576a13c53bd14f","repo":"risingwavelabs/risingwave","slug":"lance-fragment-write-task-stopped-before-accepting","errorCode":null,"errorMessage":"Lance fragment write task stopped before accepting a record batch","messagePattern":"Lance fragment write task stopped before accepting a record batch","errorType":"exception","errorClass":"SinkError::LanceDb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/lancedb.rs","lineNumber":472,"sourceCode":"    async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {\n        let record_batch = LanceDbConvert\n            .to_record_batch(self.arrow_schema.clone(), &chunk)\n            .context(\"failed to convert DataChunk to RecordBatch for LanceDB\")\n            .map_err(SinkError::LanceDb)?;\n\n        if self.fragment_write.is_none() {\n            self.fragment_write = Some(self.start_fragment_write());\n        }\n        self.fragment_write\n            .as_ref()\n            .expect(\"fragment write should be initialized\")\n            .sender\n            .as_ref()\n            .expect(\"fragment write sender should be initialized\")\n            .send(record_batch)\n            .await\n            .map_err(|_| {\n                SinkError::LanceDb(anyhow!(\n                    \"Lance fragment write task stopped before accepting a record batch\"\n                ))\n            })?;\n        Ok(())\n    }\n\n    async fn begin_epoch(&mut self, _epoch: u64) -> Result<()> {\n        Ok(())\n    }\n\n    async fn abort(&mut self) -> Result<()> {\n        let Some(mut fragment_write) = self.fragment_write.take() else {\n            return Ok(());\n        };\n\n        drop(fragment_write.sender.take());\n        match fragment_write\n            .task","sourceCodeStart":454,"sourceCodeEnd":490,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/lancedb.rs#L454-L490","documentation":"In `write_batch`, the writer sends each RecordBatch over a bounded mpsc channel to a spawned fragment-write task that calls `FileFragment::create_fragments`. A send error means the receiver side was dropped — the spawned task has already exited (with an error or panic) before accepting the batch. This is a wrapper error; the real cause is recorded in the fragment-write task's own error (\"failed to write lance data files\") or its JoinHandle.","triggerScenarios":"The `FileFragment::create_fragments` task terminates early — e.g. object-store write failure, invalid arrow schema mismatch between RecordBatch and the dataset schema, storage auth error, task panic — and the channel's receiver is dropped while `send(record_batch).await` is pending or before the next send.","commonSituations":"Transient S3/GCS errors or throttling during heavy sink writes; RisingWave schema drift vs. the target Lance table; insufficient permissions to write to the dataset directory; process memory pressure causing the write task to fail.","solutions":["Look for the preceding 'failed to write lance data files' or task panic error in the logs; that is the root cause to fix.","Verify the sink's RisingWave schema matches the target Lance table's Arrow schema (names, types, nullability).","Check object-store credentials, permissions, and rate limits for write operations on the dataset URI.","Retry/respawn the sink after transient storage issues; consider reducing batch pressure or WRITE_CHANNEL_CAPACITY-related memory usage."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Detect the dead channel and surface the root cause from the write task\nmatch sender.send(record_batch).await {\n    Ok(()) => Ok(()),\n    Err(_) => {\n        // The fragment write task already failed; join it to recover the real error\n        let root = task.await\n            .context(\"Lance fragment write task failed\")\n            .and_then(|r| r.context(\"failed to write lance data files\"));\n        Err(SinkError::LanceDb(anyhow!(\"fragment writer stopped before accepting a batch\").context(root.unwrap_err()))\n    }\n}","preventionTips":["Keep the RisingWave sink schema in sync with the target Lance table's Arrow schema","Ensure write permissions on the dataset URI's bucket/prefix before starting the sink","Watch for transient object-store throttling and add backoff in the storage layer","Always inspect the fragment-write task's JoinHandle result — the send error only signals the task died"],"tags":["lancedb","rust","async","channel","sink"],"backgroundTag":"broken-pipe","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}