What causes "IcebergSinkWriter should be initialized before barrier" in risingwavelabs/risingwave?

Ensure begin_epoch(epoch) is awaited and succeeds before calling write_batch

IcebergSinkWriter should be initialized before barrier

Error message

IcebergSinkWriter should be initialized before barrier

What it means

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.

Source

Thrown at src/connector/src/sink/iceberg/writer.rs:946

            Some(primary_key_column_names) => IcebergSinkWriterInner::build_upsert(
                &args.config,
                table,
                primary_key_column_names.clone(),
                &args.writer_param,
            )?,
            None => {
                IcebergSinkWriterInner::build_append_only(&args.config, table, &args.writer_param)?
            }
        };

        *self = IcebergSinkWriter::Initialized(inner);
        Ok(())
    }

    /// Write a stream chunk to sink
    async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {
        let Self::Initialized(inner) = self else {
            unreachable!("IcebergSinkWriter should be initialized before barrier");
        };
        inner.write_batch(chunk).await
    }

    /// Receive a barrier and mark the end of current epoch. When `is_checkpoint` is true, the sink
    /// writer should commit the current epoch.
    async fn barrier(&mut self, is_checkpoint: bool) -> Result<Option<SinkMetadata>> {
        let Self::Initialized(inner) = self else {
            unreachable!("IcebergSinkWriter should be initialized before barrier");
        };

        // Skip it if not checkpoint
        if !is_checkpoint {
            return Ok(None);
        }

        let data_files = inner
            .close()

View on GitHub (pinned to 6469eb736d)

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

Example fix

// before
writer.write_batch(chunk).await?;
// after
writer.begin_epoch(epoch).await?;
writer.write_batch(chunk).await?;

Defensive patterns

Strategy: validation

Validate before calling

assert!(matches!(writer, IcebergSinkWriter::Initialized(_)), "call begin_epoch before write_batch");

Type guard

fn is_initialized(w: &IcebergSinkWriter) -> bool { matches!(w, IcebergSinkWriter::Initialized(_)) }

Try / catch

if let IcebergSinkWriter::Initialized(inner) = &mut writer { inner.write_batch(chunk).await?; } else { return Err(anyhow!("writer not initialized")); }

Prevention

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

When it happens

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

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

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

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/9fd96134d18d763e. Report an issue: GitHub.