What causes "should get Sync response but get {:?}" in risingwavelabs/risingwave?

Inspect the debug-printed msg variant to identify which response type arrived; handle or skip that variant in the caller.

risingwavelabs/risingwave · error · RpcError

should get Sync response but get {:?}

Error message

should get Sync response but get {:?}

What it means

next_commit_response in ConnectorClient expects the sink writer stream to deliver a Commit response, skipping Batch responses, but received some other SinkWriterStreamResponse variant. The code wraps the mismatched message in an RpcError::Internal with its debug representation.

Source

Thrown at src/rpc_client/src/connector_client.rs:67

                is_checkpoint,
            })),
        })
        .await
    }
}

impl SinkWriterResponseReceiver {
    pub async fn next_commit_response(&mut self) -> Result<CommitResponse> {
        loop {
            match self.next_response().await? {
                SinkWriterStreamResponse {
                    response: Some(sink_writer_stream_response::Response::Commit(rsp)),
                } => return Ok(rsp),
                SinkWriterStreamResponse {
                    response: Some(sink_writer_stream_response::Response::Batch(_)),
                } => continue,
                msg => {
                    return Err(RpcError::Internal(anyhow!(
                        "should get Sync response but get {:?}",
                        msg
                    )));
                }
            }
        }
    }
}

impl<REQ: From<SinkWriterStreamRequest>> SinkWriterStreamHandle<REQ> {
    pub async fn write_batch(&mut self, epoch: u64, batch_id: u64, payload: Payload) -> Result<()> {
        self.request_sender
            .write_batch(epoch, batch_id, payload)
            .await
    }

    pub async fn barrier(&mut self, epoch: u64) -> Result<()> {
        self.request_sender.barrier(epoch, false).await

View on GitHub (pinned to 6469eb736d)

Solutions

Inspect the debug-printed msg variant to identify which response type arrived; handle or skip that variant in the caller.
Verify the connector node version matches the meta/compute node (compatible sink protocol).
Retry the sink operation / recreate the sink writer stream, since a closed or desynchronized stream cannot recover mid-call.

Defensive patterns

Strategy: try-catch

Try / catch

match client.commit_sink(epoch).await {
    Ok(()) => {},
    Err(e) if e.to_string().contains("should get Sync response") => {
        // recreate sink writer stream and retry once
    }
    Err(e) => return Err(e.into()),
}

Prevention

Keep connector and compute node versions in lockstep.
Use one stream per sink writer; never interleave Sync and Commit paths.
Log the full RpcError debug output when handling sink commit failures.

When it happens

Trigger: Calling commit_sink/next_commit_response while the connector sink stream is closed, delivers a Start/Barrier response out of order, or the sink endpoint returns an error/empty variant where a Commit (Sync path) was expected.

Common situations: Sink coordinator/connector endpoint version mismatch producing unexpected response variants; stream terminated by connector service crash; misordered use of the bidi stream API (expecting Sync when the writer returned StartSink).

Related errors

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