What causes "get different response epoch to commit epoch: {} {}" in risingwavelabs/risingwave?

Discard/flush stale responses from the stream before committing (drain until a response matching the current epoch).

risingwavelabs/risingwave · error · RpcError

get different response epoch to commit epoch: {} {}

Error message

get different response epoch to commit epoch: {} {}

What it means

commit in ConnectorClient received a CommitResponse whose epoch differs from the epoch being committed. The client asserts epoch equality as a safety invariant; on mismatch it returns RpcError::Internal with both epoch values.

Source

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

        self.send_request(SinkCoordinatorStreamRequest {
            request: Some(sink_coordinator_stream_request::Request::Commit(
                CommitMetadata { epoch, metadata },
            )),
        })
        .await?;
        match self.next_response().await? {
            SinkCoordinatorStreamResponse {
                response:
                    Some(sink_coordinator_stream_response::Response::Commit(
                        sink_coordinator_stream_response::CommitResponse {
                            epoch: response_epoch,
                        },
                    )),
            } => {
                if epoch == response_epoch {
                    Ok(())
                } else {
                    Err(RpcError::Internal(anyhow!(
                        "get different response epoch to commit epoch: {} {}",
                        epoch,
                        response_epoch
                    )))
                }
            }
            msg => Err(RpcError::Internal(anyhow!(
                "should get Commit response but get {:?}",
                msg
            ))),
        }
    }
}

View on GitHub (pinned to 6469eb736d)

Solutions

Discard/flush stale responses from the stream before committing (drain until a response matching the current epoch).
Do not share one ConnectorClient/sink stream across concurrent epoch commits; serialize commits per stream.
Log both epoch values and recreate the sink writer; a mismatch indicates the stream is out of sync and cannot be trusted.

Example fix

// before
if epoch == response_epoch { Ok(()) } else { Err(...) }
// after: skip stale responses first
loop {
    match next_response().await? {
        SinkWriterStreamResponse { response: Some(Response::Commit(rsp)) } => {
            if rsp.epoch == epoch { break Ok(()); } // else: stale, keep draining
        }
        _ => continue,
    }
}

Defensive patterns

Strategy: retry

Try / catch

loop {
    match client.commit(epoch).await {
        Ok(()) => break,
        Err(e) if e.to_string().contains("different response epoch") => {
            // stale response in stream: recreate stream and retry
            handle = recreate_sink_stream().await?;
        }
        Err(e) => return Err(e.into()),
    }
}

Prevention

Never share one sink stream across concurrent epoch commits.
After any commit timeout, recreate the stream instead of reusing it.
Ensure epochs strictly increase per stream session.

When it happens

Trigger: Calling commit(epoch) while the connector sink stream has a stale Commit response queued from a previous (aborted or retried) commit, so response_epoch != epoch.

Common situations: Retried commits after a timeout where the old response is still buffered in the stream; concurrent commit calls sharing one sink writer stream; epoch reuse after barrier failure.

Related errors

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