risingwavelabs/risingwave · error · MetaError

old sink job {} not found in barrier state

Error message

old sink job {} not found in barrier state

What it means

During a replace-sink apply, the old sink job id being replaced was not found in the barrier state's job registry (post_apply_remove_job returned None), so the applier cannot collect its dropped actors and bails out.

Source

Thrown at src/meta/src/barrier/checkpoint/state.rs:954

                        }),
                );
                if let CreateStreamingJobType::SinkIntoTable(ref ctx) = job_type {
                    let downstream_fragment_id = ctx.new_sink_downstream.downstream_fragment_id;
                    self.database_info.pre_apply_add_node_upstream(
                        downstream_fragment_id,
                        &PbUpstreamSinkInfo {
                            upstream_fragment_id: ctx.sink_fragment_id,
                            sink_output_schema: ctx.sink_output_fields.clone(),
                            project_exprs: ctx.project_exprs.clone(),
                        },
                    );
                }

                let (table_ids, node_actors) = self.collect_base_info();
                let dropped_actors = if let Some(old_sink_job_id) = old_sink_job_id {
                    let Some(job) = self.database_info.post_apply_remove_job(old_sink_job_id)
                    else {
                        bail!(
                            "old sink job {} not found in barrier state",
                            old_sink_job_id
                        );
                    };
                    job.fragment_infos
                        .values()
                        .flat_map(|fragment| fragment.actors.keys().copied())
                        .collect()
                } else {
                    vec![]
                };

                // Actors to create
                let actors_to_create = Some(Command::create_streaming_job_actors_to_create(
                    &info,
                    &mut edges,
                    &actors.stream_actors,
                    &actors.actor_location,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the old sink still exists before issuing the replace
2. Retry the operation; if the sink is gone, create a new sink instead
3. Check for concurrent DDL races and re-run after the conflicting command completes
4. Report a bug if it reproduces on a stable barrier state

if barrier_state.database_info.get_job(old_sink_job_id).is_none() {
    return Err("old sink already removed; create new sink instead");
}

fn old_sink_exists(state: &BarrierState, id: JobId) -> bool {
    state.database_info.get_job(id).map(|j| j.is_sink()).unwrap_or(false)
}

match replace_sink(...).await {
    Err(e) if e.contains("not found in barrier state") => create_sink_fresh().await,
    other => other,
}

When it happens

Trigger: Applying a replace-sink command via handle_new_barrier where old_sink_job_id references a sink that no longer exists in database_info (already dropped, finished, or failed).

Common situations: Concurrent DROP SINK and replace-sink; retry of a replace after a failure that already removed the old job; stale metadata after recovery.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

• replace sink must not use snapshot backfill
• should get metadata on checkpoint barrier
• since_timestamp epoch has not been resolved for snapshot bac
• cannot create batch refresh job while database barrier is pa
• no table changelog found for upstream table {} when resolvin

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