{"record":{"id":"c9b35de96c6ba1ce","repo":"risingwavelabs/risingwave","slug":"should-get-metadata-on-checkpoint-barrier","errorCode":null,"errorMessage":"should get metadata on checkpoint barrier","messagePattern":"should get metadata on checkpoint barrier","errorType":"exception","errorClass":"SinkError::Coordinator","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/coordinate.rs","lineNumber":215,"sourceCode":"                    schema_change,\n                } => {\n                    let prev_epoch = match state {\n                        LogConsumerState::EpochBegun { curr_epoch } => curr_epoch,\n                        _ => unreachable!(\"epoch must have begun before handling barrier\"),\n                    };\n                    if is_checkpoint {\n                        current_checkpoint += 1;\n                        if current_checkpoint >= commit_checkpoint_interval.get()\n                            || should_force_commit_on_checkpoint_barrier(\n                                new_vnode_bitmap.is_some(),\n                                is_stop,\n                                schema_change.is_some(),\n                            )\n                        {\n                            let start_time = Instant::now();\n                            let metadata = sink_writer.barrier(true).await?;\n                            let metadata = metadata.ok_or_else(|| {\n                                SinkError::Coordinator(anyhow!(\n                                    \"should get metadata on checkpoint barrier\"\n                                ))\n                            })?;\n                            if schema_change.is_some() {\n                                tracing::info!(\n                                    sink_id = %self.param.sink_id,\n                                    ?schema_change,\n                                    \"schema change received for coordinated log sinker\"\n                                );\n                                assert!(\n                                    is_stop,\n                                    \"schema change should stop current sink for sink {}\",\n                                    self.param.sink_id\n                                );\n                            }\n                            coordinator_stream_handle\n                                .commit(epoch, metadata, schema_change)\n                                .await?;","sourceCodeStart":197,"sourceCodeEnd":233,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/coordinate.rs#L197-L233","documentation":"On a checkpoint barrier the coordinator calls sink_writer.barrier(true), which must return metadata describing the sink's state. A None result means the writer did not produce metadata for a forced checkpoint, violating the coordinator protocol, so it wraps this as a SinkError::Coordinator.","triggerScenarios":"consume_log_and_sink processes a checkpoint barrier (with or without schema change) and `sink_writer.barrier(true).await` resolves to None.","commonSituations":"Using a sink writer implementation that does not support checkpoint metadata; bugs in a custom/dynamic sink writer where the checkpoint path returns None instead of Some(metadata).","solutions":["Inspect the sink writer's barrier(true) implementation to ensure it always returns Some(metadata) on forced checkpoints.","Verify the sink connector type supports coordination metadata (not all sinks do).","Upgrade RisingWave / connector crate if this is a known sink implementation bug."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"let metadata = sink_writer.barrier(true).await?\n    .ok_or_else(|| SinkError::Coordinator(anyhow!(\"should get metadata on checkpoint barrier\")))?;","preventionTips":["Use sink writers known to implement checkpoint metadata for coordinated sinks.","Add unit tests asserting barrier(true) returns Some(metadata) for every sink implementation."],"tags":["rust","risingwave","sink","barrier"],"backgroundTag":"unexpected-response-shape","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"}