{"record":{"id":"dd37fd9f23c62e7f","repo":"risingwavelabs/risingwave","slug":"get-unexpected-response","errorCode":null,"errorMessage":"get unexpected response: {:?}","messagePattern":"get unexpected response: (.+?)","errorType":"exception","errorClass":"SinkError::Remote","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/remote.rs","lineNumber":446,"sourceCode":"                                    Some(sink_writer_stream_response::Response::Commit(\n                                        sink_writer_stream_response::CommitResponse {\n                                            epoch,\n                                            metadata,\n                                        },\n                                    )),\n                            } => {\n                                if let Some(metadata) = metadata {\n                                    warn!(\"get unexpected non-empty metadata: {:?}\", metadata);\n                                }\n                                truncate_matched_offset(\n                                    &mut sent_offset_queue,\n                                    TruncateOffset::Barrier { epoch },\n                                    &mut log_reader,\n                                    &sink_writer_metrics,\n                                )?;\n                            }\n                            response => {\n                                return Err(SinkError::Remote(anyhow!(\n                                    \"get unexpected response: {:?}\",\n                                    response\n                                )));\n                            }\n                        }\n                    }\n                    futures::future::Either::Right(result) => {\n                        let (epoch, item): (u64, LogStoreReadItem) = result?;\n\n                        match item {\n                            LogStoreReadItem::StreamChunk { chunk, chunk_id } => {\n                                let offset = TruncateOffset::Chunk { epoch, chunk_id };\n                                if let Some(prev_offset) = &prev_offset {\n                                    prev_offset.check_next_offset(offset)?;\n                                }\n                                let cardinality = chunk.cardinality();\n                                sink_writer_metrics\n                                    .connector_sink_rows_received","sourceCodeStart":428,"sourceCodeEnd":464,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/remote.rs#L428-L464","documentation":"consume_log_and_sink matches on the response variants it expects (e.g. Commit, TruncateOffset, BeginEpoch/WriteBatch responses). Any other response variant arriving on the stream is a protocol violation and is reported with its Debug representation.","triggerScenarios":"A JniSinkWriterStreamResponse arrives that doesn't match the handled variants in the current state of the loop - e.g. a BeginEpoch or Chunk response arriving when the code expects Commit/Truncate handling.","commonSituations":"Version skew between the stream engine and the embedded JVM connector (new response type added); a corrupted/misordered stream after a reconnect; duplicated epoch-begin responses after recovery.","solutions":["Note the Debug-printed response variant; check for a version mismatch between risingwave and the connector/JVM side.","Recover the sink from the last barrier to reset the response stream state.","If a new response variant was added upstream, upgrade RisingWave so the handler recognizes it."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match response {\n    Commit{..} | TruncateOffset{..} => handle_expected(response),\n    other => {\n        log::error!(\"unexpected sink response: {other:?}\");\n        // check version skew, then recover from last barrier\n        request_recovery_from_last_barrier();\n    }\n}","preventionTips":["Keep RisingWave and the embedded connector versions in lockstep","Handle every enum variant explicitly so new variants fail at compile time","Log full response Debug output to speed up version-skew diagnosis"],"tags":["sink","protocol","unexpected-response","version-skew"],"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-23T08:17:48.524Z"}