{"record":{"id":"cb9b88239ec3d3d3","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-writer-replacement-input-close","errorCode":null,"errorMessage":"iceberg pk-index writer {} replacement input closed before its initial barrier for task {}","messagePattern":"iceberg pk-index writer (.+?) replacement input closed before its initial barrier for task (.+?)","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"critical","filePath":"src/stream/src/executor/iceberg_with_pk_index/writer.rs","lineNumber":554,"sourceCode":"                ),\n                Message::Watermark(_) => bail!(\n                    \"iceberg pk-index writer {} received watermark from replacement input before its initial barrier for task {}\",\n                    self.sink_id,\n                    task_id\n                ),\n                Message::Barrier(barrier) => {\n                    self.validate_aligned_barriers(&barrier, &expected)?;\n                    #[for_await]\n                    for msg in self.checkpoint_barrier(barrier) {\n                        yield msg?;\n                    }\n                    self.mode = WriterInputMode::Normal;\n                    return Ok(());\n                }\n            }\n        }\n\n        bail!(\n            \"iceberg pk-index writer {} replacement input closed before its initial barrier for task {}\",\n            self.sink_id,\n            task_id\n        );\n    }\n}\n\nimpl<S, W> Execute for WriterExecutor<S, W>\nwhere\n    S: StateStore,\n    W: IcebergWriter,\n{\n    fn execute(self: Box<Self>) -> BoxedMessageStream {\n        self.execute_inner().boxed()\n    }\n}\n\nasync fn next_msg(input: &mut BoxedMessageStream) -> StreamExecutorResult<Message> {","sourceCodeStart":536,"sourceCodeEnd":572,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/writer.rs#L536-L572","documentation":"The Iceberg sink writer executor (pk-index variant) requires its replacement input to deliver an initial barrier before producing any other message, so it can align with the barrier expected from the previous phase of a compaction/resolution task. When the replacement input stream terminates (returns None) without ever yielding that barrier, the writer cannot complete the mode transition back to WriterInputMode::Normal and bails. This is a stream protocol/invariant violation: inputs are expected to stay alive until at least one barrier is delivered.","triggerScenarios":"execute_aligning_replacement_input is entered with an expected barrier from the resolving phase; the replacement (switched-to) input then ends — upstream actor terminated, failed, or the fragment scheduler stopped the upstream — before emitting its first Message::Barrier. Any Chunk or Watermark before the barrier also bails (different message), but stream end without a barrier triggers exactly this error.","commonSituations":"Upstream actor crash or failure during an Iceberg compaction task; cluster scaling/rescaling that removes the replacement upstream mid-task; stream graph rewrite (e.g. MV schema change) replacing inputs while the writer is mid-compaction; internal scheduler bugs that stop an upstream before barrier alignment completes.","solutions":["Inspect the failing upstream actor's logs (same fragment/task) for an earlier error or panic that ended its stream and fix that root cause first.","Retry the stream job/recovery — barrier-based recovery restarts the fragment and rebuilds the compaction state.","Check whether a rescale or schema-change was applied concurrently with the compaction task; re-run the operation without overlapping them.","If reproducible with no upstream error, file a bug with the fragment graph and task_id — this indicates a scheduler/writer protocol bug."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match writer.run().await {\n    Err(e) if e.to_string().contains(\"replacement input closed before its initial barrier\") => {\n        // log task_id/sink_id, trigger fragment recovery/restart of the compaction task\n        tracing::error!(%e, \"iceberg writer replacement input died pre-barrier; recovering\");\n        recover_fragment().await?;\n    }\n    other => other?,\n}","preventionTips":["Avoid rescaling or schema changes while an Iceberg compaction task is in the AligningReplacementInput phase.","Monitor upstream actor liveness/health so failures are caught before the writer observes a dead input.","Enable debug logging around sink/compaction barrier alignment to catch protocol anomalies early.","Ensure barrier delivery is not disabled or throttled in test/custom deployments."],"tags":["streaming","internal-invariant","barrier","iceberg","rust"],"backgroundTag":"internal-invariant-violation","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"}