{"record":{"id":"9fd96134d18d763e","repo":"risingwavelabs/risingwave","slug":"icebergsinkwriter-should-be-initialized-before-bar","errorCode":null,"errorMessage":"IcebergSinkWriter should be initialized before barrier","messagePattern":"IcebergSinkWriter should be initialized before barrier","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"src/connector/src/sink/iceberg/writer.rs","lineNumber":946,"sourceCode":"            Some(primary_key_column_names) => IcebergSinkWriterInner::build_upsert(\n                &args.config,\n                table,\n                primary_key_column_names.clone(),\n                &args.writer_param,\n            )?,\n            None => {\n                IcebergSinkWriterInner::build_append_only(&args.config, table, &args.writer_param)?\n            }\n        };\n\n        *self = IcebergSinkWriter::Initialized(inner);\n        Ok(())\n    }\n\n    /// Write a stream chunk to sink\n    async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {\n        let Self::Initialized(inner) = self else {\n            unreachable!(\"IcebergSinkWriter should be initialized before barrier\");\n        };\n        inner.write_batch(chunk).await\n    }\n\n    /// Receive a barrier and mark the end of current epoch. When `is_checkpoint` is true, the sink\n    /// writer should commit the current epoch.\n    async fn barrier(&mut self, is_checkpoint: bool) -> Result<Option<SinkMetadata>> {\n        let Self::Initialized(inner) = self else {\n            unreachable!(\"IcebergSinkWriter should be initialized before barrier\");\n        };\n\n        // Skip it if not checkpoint\n        if !is_checkpoint {\n            return Ok(None);\n        }\n\n        let data_files = inner\n            .close()","sourceCodeStart":928,"sourceCodeEnd":964,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/writer.rs#L928-L964","documentation":"IcebergSinkWriter is a state machine with Created/Initialized states. write_batch requires the writer to be in the Initialized state (i.e. begin_epoch has run and built the inner writer); if not, an unreachable!() panics because it indicates a protocol violation by the stream executor. The library treats this as an internal invariant, not a user-facing error.","triggerScenarios":"Calling SinkWriter::write_batch on an IcebergSinkWriter that is still in the Created state, i.e. before begin_epoch() completed successfully or after it was skipped/failed.","commonSituations":"Custom executor wiring that streams chunks before a barrier/epoch start; begin_epoch failing silently upstream; tests invoking write_batch directly on a fresh IcebergSinkWriter.","solutions":["Ensure begin_epoch(epoch) is awaited and succeeds before calling write_batch","Check that the stream executor does not replay chunks before the first barrier initializes the writer","If begin_epoch can fail, surface that error instead of proceeding to write_batch"],"exampleFix":"// before\nwriter.write_batch(chunk).await?;\n// after\nwriter.begin_epoch(epoch).await?;\nwriter.write_batch(chunk).await?;","handlingStrategy":"validation","validationCode":"assert!(matches!(writer, IcebergSinkWriter::Initialized(_)), \"call begin_epoch before write_batch\");","typeGuard":"fn is_initialized(w: &IcebergSinkWriter) -> bool { matches!(w, IcebergSinkWriter::Initialized(_)) }","tryCatchPattern":"if let IcebergSinkWriter::Initialized(inner) = &mut writer { inner.write_batch(chunk).await?; } else { return Err(anyhow!(\"writer not initialized\")); }","preventionTips":["Always drive the SinkWriter lifecycle in order: begin_epoch -> write_batch -> barrier","Check the Result of begin_epoch before streaming data","Add an integration test that replays executor ordering edge cases"],"tags":["rust","iceberg","sink","state-machine","panic"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}