From f6026139f1a57d9d7e8f10ef979932cf625bd5f4 Mon Sep 17 00:00:00 2001 From: Gunther Xing Date: Sun, 26 Jul 2026 00:46:42 +0800 Subject: [PATCH] feat(common): add serialized error payloads for data-bearing error variants Server now serializes the fields of data-bearing error variants (e.g. PartitionNotFound, StreamIdNotFound) into a binary payload and sends it alongside the error code. Client transports (QUIC, TCP, WebSocket, VSR) read this payload and reconstruct the full error, enabling contextual error messages on the client side. The Python SDK e2e test is updated to match the new dynamic IDs. --- core/common/src/error/iggy_error.rs | 342 +++++++++++++++++- core/sdk/src/quic/quic_client.rs | 13 +- core/sdk/src/tcp/tcp_client.rs | 46 ++- core/sdk/src/vsr.rs | 67 +++- core/sdk/src/websocket/websocket_client.rs | 7 +- core/server/src/sender/mod.rs | 3 +- .../python/tests/test_message_operations.py | 4 +- 7 files changed, 473 insertions(+), 9 deletions(-) diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs index 4d9d8c0bf7..80117c8bd9 100644 --- a/core/common/src/error/iggy_error.rs +++ b/core/common/src/error/iggy_error.rs @@ -15,9 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::Identifier; use crate::utils::topic_size::MaxTopicSize; +use crate::{IdKind, Identifier}; use crate::{IggyMessage, utils::byte_size::IggyByteSize}; +use std::mem::size_of; use std::sync::Arc; use strum::{EnumDiscriminants, FromRepr, IntoStaticStr}; use thiserror::Error; @@ -533,6 +534,152 @@ pub enum IggyError { IncompatibleProtocolVersion(u32, u32, u32) = 14003, } +/// Trait for serializing/deserializing individual [`IggyError`] payload fields. +trait ErrorPayloadField: Sized { + fn write(&self, buf: &mut Vec); + fn read(buf: &[u8], pos: usize) -> (Self, usize); +} + +impl ErrorPayloadField for usize { + fn write(&self, buf: &mut Vec) { + buf.extend_from_slice(&(*self as u64).to_le_bytes()); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + let bytes: [u8; size_of::()] = buf[pos..pos + size_of::()] + .try_into() + .expect("invalid usize payload"); + (u64::from_le_bytes(bytes) as usize, pos + size_of::()) + } +} + +impl ErrorPayloadField for u32 { + fn write(&self, buf: &mut Vec) { + buf.extend_from_slice(&self.to_le_bytes()); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + let bytes: [u8; size_of::()] = buf[pos..pos + size_of::()] + .try_into() + .expect("invalid u32 payload"); + (Self::from_le_bytes(bytes), pos + size_of::()) + } +} + +impl ErrorPayloadField for u64 { + fn write(&self, buf: &mut Vec) { + buf.extend_from_slice(&self.to_le_bytes()); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + let bytes: [u8; size_of::()] = buf[pos..pos + size_of::()] + .try_into() + .expect("invalid u64 payload"); + (Self::from_le_bytes(bytes), pos + size_of::()) + } +} + +impl ErrorPayloadField for u16 { + fn write(&self, buf: &mut Vec) { + buf.extend_from_slice(&self.to_le_bytes()); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + let bytes: [u8; size_of::()] = buf[pos..pos + size_of::()] + .try_into() + .expect("invalid u16 payload"); + (Self::from_le_bytes(bytes), pos + size_of::()) + } +} + +impl ErrorPayloadField for u8 { + fn write(&self, buf: &mut Vec) { + buf.push(*self); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + (buf[pos], pos + 1) + } +} + +impl ErrorPayloadField for Identifier { + fn write(&self, buf: &mut Vec) { + buf.push(self.kind as u8); + buf.push(self.length); + buf.extend_from_slice(&self.value); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + let kind = match buf[pos] { + 0 => IdKind::Numeric, + _ => IdKind::String, + }; + let length = buf[pos + 1] as usize; + let value = buf[pos + 2..pos + 2 + length].to_vec(); + ( + Identifier { + kind, + length: length as u8, + value, + }, + pos + 2 + length, + ) + } +} + +impl ErrorPayloadField for String { + fn write(&self, buf: &mut Vec) { + let len = self.len() as u32; + buf.extend_from_slice(&len.to_le_bytes()); + buf.extend_from_slice(self.as_bytes()); + } + fn read(buf: &[u8], pos: usize) -> (Self, usize) { + let len = u32::from_le_bytes( + buf[pos..pos + size_of::()] + .try_into() + .expect("invalid string length payload"), + ) as usize; + let data_start = pos + size_of::(); + let data = String::from_utf8_lossy(&buf[data_start..data_start + len]).into_owned(); + (data, data_start + len) + } +} + +macro_rules! implement_error_payload { + ($($variant:ident($($field:ident: $ty:ty),* $(,)?)),* $(,)?) => { + /// Serialize the fields of a data-bearing error variant into a binary + /// payload. Returns an empty vector for unit variants and unhandled + /// variants. + pub fn write_payload(&self) -> Vec { + let mut buf = Vec::new(); + match self { + $(IggyError::$variant($($field),*) => { + $(ErrorPayloadField::write($field, &mut buf);)* + })* + _ => {} + } + buf + } + + /// Reconstruct an [`IggyError`] from a server-provided code and a + /// structured binary payload. Falls back to [`IggyError::from_code`] + /// when the payload is empty or the code maps to a unit variant or + /// an unrecognized variant. + pub fn from_code_with_payload(code: u32, payload: &[u8]) -> Self { + if payload.is_empty() { + return IggyError::from_code(code); + } + let Some(disc) = IggyErrorDiscriminants::from_repr(code) else { + return IggyError::from_code(code); + }; + let mut cursor: usize = 0; + match disc { + $(IggyErrorDiscriminants::$variant => { + $(let ($field, off) = <$ty as ErrorPayloadField>::read(payload, cursor); + cursor = off;)* + let _ = cursor; + IggyError::$variant($($field),*) + })* + _ => IggyError::from_code(code), + } + } + }; +} + impl IggyError { pub fn as_code(&self) -> u32 { // SAFETY: SdkError specifies #[repr(u32)] representation. @@ -553,6 +700,108 @@ impl IggyError { .map(|discriminant| discriminant.into()) .unwrap_or("unknown error code") } + + implement_error_payload! { + PartitionNotFound(partition_id: usize, topic_id: Identifier, stream_id: Identifier), + StreamIdNotFound(stream_id: Identifier), + TopicIdNotFound(stream_id: Identifier, topic_id: Identifier), + InvalidOffset(offset: u64), + ClientNotFound(client_id: u32), + CannotDeleteUser(user_id: u32), + CannotChangePermissions(user_id: u32), + ConsumerOffsetNotFound(offset: usize), + NoPartitions(topic_id: Identifier, stream_id: Identifier), + TopicFull(topic_id: Identifier, stream_id: Identifier), + ConsumerGroupIdNotFound(group_id: Identifier, topic_id: Identifier), + ConsumerGroupMemberNotFound(client_id: u32, group_id: Identifier, topic_id: Identifier), + ConsumerGroupPartitionNotOwned(client_id: u32, partition_id: u32), + CannotDeletePartitionDirectory(partition_id: usize, stream_id: usize, topic_id: usize), + CannotCreateConsumerGroupInfo(id: usize, topic_id: Identifier, stream_id: Identifier), + CannotDeleteConsumerGroupInfo(id: usize, topic_id: Identifier, stream_id: Identifier), + ConsumerGroupNameNotFound(name: String, topic_id: Identifier), + ConsumerGroupNameAlreadyExists(name: String, topic_id: Identifier), + InvalidReservedField(value: u64), + // Single-field String variants. + InvalidVersion(info: String), + CannotCreateBaseDirectory(path: String), + CannotCreateRuntimeDirectory(path: String), + CannotRemoveRuntimeDirectory(path: String), + CannotCreateStateDirectory(path: String), + CannotOpenDatabase(path: String), + ResourceNotFound(key: String), + CannotCloseWebSocketConnection(reason: String), + HttpError(info: String), + InvalidApiUrl(url: String), + InvalidJwtAlgorithm(alg: String), + CannotFetchJwks(url: String), + CannotParseHeaderKind(info: String), + CannotCreateStreamsDirectory(path: String), + StreamNameNotFound(name: String), + StreamDirectoryNotFound(path: String), + StreamNameAlreadyExists(name: String), + TopicDirectoryNotFound(path: String), + CannotDeleteConsumerOffsetsDirectory(path: String), + CannotDeleteConsumerOffsetFile(path: String), + CannotCreateConsumerOffsetsDirectory(path: String), + CannotReadConsumerOffsets(path: String), + CannotOpenConsumerOffsetsFile(path: String), + CannotCreateSegmentLogFile(path: String), + CannotCreateSegmentIndexFile(path: String), + CannotCreateSegmentTimeIndexFile(path: String), + CommandLengthError(info: String), + CannotBindToSocket(info: String), + IoError(info: String), + // Remaining compatible data-bearing variants. + InvalidStateEntryChecksum(a: u64, b: u64, c: u64), + InvalidIpAddress(ip: String, info: String), + HttpResponseError(status: u16, body: String), + PersonalAccessTokenAlreadyExists(token: String, user_id: u32), + PersonalAccessTokensLimitReached(user_id: u32, limit: u32), + PersonalAccessTokenExpired(token: String, user_id: u32), + CannotCreateStream(s: u32), + CannotCreateStreamDirectory(stream_id: u32, path: String), + CannotCreateStreamInfo(s: u32), + CannotUpdateStreamInfo(s: u32), + CannotOpenStreamInfo(s: u32), + CannotReadStreamInfo(s: u32), + CannotDeleteStream(s: u32), + CannotDeleteStreamDirectory(s: u32), + CannotCreateTopicsDirectory(stream_id: Identifier, path: String), + CannotCreateTopicDirectory(stream_id: usize, topic_id: usize, path: String), + CannotCreateTopic(stream_id: u32, topic_id: u32), + CannotCreateTopicInfo(stream_id: u32, topic_id: u32), + CannotDeleteTopic(stream_id: u32, topic_id: u32), + CannotDeleteTopicDirectory(stream_id: u32, topic_id: u32, path: String), + CannotOpenTopicInfo(stream_id: u32, topic_id: u32), + CannotReadTopicInfo(stream_id: u32, topic_id: u32), + CannotReadTopics(stream_id: u32), + CannotUpdateTopicInfo(stream_id: u32, topic_id: u32), + CannotCreatePartition(partition_id: usize, stream_id: usize, topic_id: usize), + CannotCreatePartitionDirectory(partition_id: usize, stream_id: usize, topic_id: usize), + CannotCreatePartitionsDirectory(stream_id: usize, topic_id: usize), + CannotDeletePartition(partition_id: usize, stream_id: Identifier, topic_id: Identifier), + InvalidBatchChecksum(a: u64, b: u64, c: u64), + InvalidHeaderKind(kind: u8), + InvalidIndexesByteSize(size: u32), + InvalidIndexesCount(count: u32, max: u32), + InvalidMessageChecksum(a: u64, b: u64, c: u64), + InvalidMessageTimestampDelta(delta: u64), + InvalidMessagesSize(size: u32, max: u32), + InvalidSegmentSize(size: u64), + InvalidSegmentsCount(count: u32), + InvalidSession(session: u64), + MissingIndex(index: u32), + NonZeroOffset(offset: u64, partition_id: u32), + NonZeroTimestamp(timestamp: u64, partition_id: u32), + NotResolvedConsumer(consumer_id: Identifier), + SegmentClosed(offset: u64, partition_id: usize), + ShardNotFound(shard_id: usize, topic_id: usize, stream_id: usize), + TimestampOutOfRange(timestamp: u64), + TooSmallMessage(size: u32, min: u32), + TopicNameAlreadyExists(name: String, topic_id: Identifier), + TopicNameNotFound(name: String, topic_id: String), + UnknownReplicatedCommand(code: u32), + } } impl PartialEq for IggyError { @@ -594,4 +843,95 @@ mod tests { IggyError::from_code_as_string(GROUP_NAME_ERROR_CODE) ) } + + #[test] + fn from_code_reconstructs_unit_variant_preserving_code() { + let err = IggyError::from_code(IggyError::Unauthorized.as_code()); + assert_eq!(err.as_code(), IggyError::Unauthorized.as_code()); + assert!(matches!(err, IggyError::Unauthorized)); + } + + #[test] + fn from_code_fills_data_bearing_variant_with_default_fields() { + let code: u32 = 3007; // PartitionNotFound + let err = IggyError::from_code(code); + assert_eq!(err.as_code(), code); + // from_repr uses Default for data fields: partition 0, zeroed identifiers. + assert!(matches!(err, IggyError::PartitionNotFound(..))); + let error_string = err.to_string(); + assert!(error_string.contains("Partition with ID: 0")); + } + + #[test] + fn payload_round_trip_reconstructs_partition_not_found_with_real_fields() { + let original = IggyError::PartitionNotFound( + 5, + crate::Identifier::numeric(1).unwrap(), + crate::Identifier::numeric(2).unwrap(), + ); + let payload = original.write_payload(); + assert!(!payload.is_empty()); + let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload); + assert_eq!(reconstructed.as_code(), original.as_code()); + assert!(matches!(&reconstructed, IggyError::PartitionNotFound( + 5, + topic_id, + stream_id + ) if topic_id.value == 1u32.to_le_bytes().to_vec() + && stream_id.value == 2u32.to_le_bytes().to_vec())); + } + + #[test] + fn payload_round_trip_reconstructs_stream_id_not_found_with_real_fields() { + let original = IggyError::StreamIdNotFound(crate::Identifier::numeric(42).unwrap()); + let payload = original.write_payload(); + assert!(!payload.is_empty()); + let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload); + assert_eq!(reconstructed.as_code(), original.as_code()); + assert!( + matches!(&reconstructed, IggyError::StreamIdNotFound(id) if id.value == 42u32.to_le_bytes().to_vec()) + ); + } + + #[test] + fn payload_round_trip_reconstructs_invalid_offset_with_real_fields() { + let original = IggyError::InvalidOffset(123456); + let payload = original.write_payload(); + assert!(!payload.is_empty()); + let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload); + assert_eq!(reconstructed.as_code(), original.as_code()); + assert!(matches!(reconstructed, IggyError::InvalidOffset(123456))); + } + + #[test] + fn payload_round_trip_reconstructs_client_not_found_with_real_fields() { + let original = IggyError::ClientNotFound(99); + let payload = original.write_payload(); + assert!(!payload.is_empty()); + let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload); + assert_eq!(reconstructed.as_code(), original.as_code()); + assert!(matches!(reconstructed, IggyError::ClientNotFound(99))); + } + + #[test] + fn payload_round_trip_unit_variant_produces_empty_payload() { + let original = IggyError::Unauthorized; + let payload = original.write_payload(); + assert!(payload.is_empty()); + let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload); + assert_eq!(reconstructed, original); + } + + #[test] + fn from_code_with_payload_falls_back_to_from_code_when_payload_is_empty() { + let code = IggyError::Unauthenticated.as_code(); + let err = IggyError::from_code_with_payload(code, &[]); + assert_eq!(err, IggyError::Unauthenticated); + } + + #[test] + fn from_code_unknown_code_falls_back_to_error_variant() { + let err = IggyError::from_code(0xDEAD_BEEF); + assert_eq!(err, IggyError::Error); + } } diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 9a731f6082..2c39862b6b 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -354,13 +354,24 @@ impl QuicClient { .map_err(|_| IggyError::InvalidNumberEncoding)?, ); if status != 0 { + let error_length = u32::from_le_bytes( + buffer[4..RESPONSE_INITIAL_BYTES_LENGTH] + .try_into() + .map_err(|_| IggyError::InvalidNumberEncoding)?, + ); + let body = if error_length > 0 { + let body_end = RESPONSE_INITIAL_BYTES_LENGTH + error_length as usize; + &buffer[RESPONSE_INITIAL_BYTES_LENGTH..body_end.min(buffer.len())] + } else { + &[] + }; error!( "Received an invalid response with status: {} ({}).", status, IggyError::from_code_as_string(status) ); - return Err(IggyError::from_code(status)); + return Err(IggyError::from_code_with_payload(status, body)); } let length = u32::from_le_bytes( diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index f827b9f55b..f2e1f06e57 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -327,6 +327,11 @@ impl TcpClient { stream: &mut ConnectionStreamKind, ) -> Result { if status != 0 { + let mut body = BytesMut::new(); + if length > 0 { + body.put_bytes(0, length as usize); + stream.read(&mut body).await?; + } // TEMP: See https://github.com/apache/iggy/pull/604 for context. if status == IggyErrorDiscriminants::TopicNameAlreadyExists as u32 || status == IggyErrorDiscriminants::StreamNameAlreadyExists as u32 @@ -347,7 +352,7 @@ impl TcpClient { ); } - return Err(IggyError::from_code(status)); + return Err(IggyError::from_code_with_payload(status, &body)); } trace!("Status: OK. Response length: {}", length); @@ -1334,6 +1339,45 @@ mod tests { assert!(tcp_client.is_err()); } + #[cfg(not(feature = "vsr"))] + #[tokio::test] + async fn should_reconstruct_data_bearing_error_from_payload_when_body_is_non_empty() { + let code: u32 = 3007; // PartitionNotFound + // partition_id=5, topic_id=1 (numeric), stream_id=2 (numeric) + let mut payload = Vec::new(); + payload.extend_from_slice(&5u64.to_le_bytes()); // partition_id as u64 + payload.push(0); // IdKind::Numeric + payload.push(4); // length + payload.extend_from_slice(&1u32.to_le_bytes()); // topic_id = 1 + payload.push(0); // IdKind::Numeric + payload.push(4); // length + payload.extend_from_slice(&2u32.to_le_bytes()); // stream_id = 2 + let mut stream = make_dummy_stream(&payload).await; + let result = TcpClient::handle_response(code, payload.len() as u32, &mut stream).await; + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.as_code(), code); + assert!(matches!(&err, IggyError::PartitionNotFound( + 5, + topic_id, + stream_id + ) if topic_id.value == 1u32.to_le_bytes().to_vec() + && stream_id.value == 2u32.to_le_bytes().to_vec())); + } + + #[cfg(not(feature = "vsr"))] + #[tokio::test] + async fn should_fallback_to_default_error_when_error_body_is_empty() { + let mut stream = make_dummy_stream(&[]).await; + let code: u32 = 3007; // PartitionNotFound + let result = TcpClient::handle_response(code, 0, &mut stream).await; + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.as_code(), code); + // Fallback: old server sends no body, so from_code produces Default fields. + assert!(matches!(err, IggyError::PartitionNotFound(..))); + } + #[cfg(not(feature = "vsr"))] #[tokio::test] async fn should_return_ok_when_status_is_zero() { diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs index d1e9ab859c..0bb674de38 100644 --- a/core/sdk/src/vsr.rs +++ b/core/sdk/src/vsr.rs @@ -184,7 +184,12 @@ pub(crate) fn decode_response(response: Bytes) -> Result { return Err(IggyError::InvalidCommand); } if let Some(error) = read_reply_status(header_bytes) { - return Err(error); + let body = if total_size > HEADER_SIZE { + &response[HEADER_SIZE..total_size.min(response.len())] + } else { + &[] + }; + return Err(IggyError::from_code_with_payload(error.as_code(), body)); } let operation = read_operation(header_bytes)?; split_metadata_result(operation, response.slice(HEADER_SIZE..total_size)) @@ -214,7 +219,12 @@ pub(crate) fn decode_response_split( return Err(IggyError::InvalidCommand); } if let Some(error) = read_reply_status(header_bytes) { - return Err(error); + let payload = if expected_body > 0 { + &body[..expected_body.min(body.len())] + } else { + &[] + }; + return Err(IggyError::from_code_with_payload(error.as_code(), payload)); } let operation = read_operation(header_bytes)?; split_metadata_result(operation, body.slice(..expected_body)) @@ -593,6 +603,59 @@ mod tests { assert!(matches!(result, Err(IggyError::Unauthorized))); } + #[test] + fn reply_with_nonzero_data_bearing_code_reconstructs_from_payload() { + let code = + IggyError::PartitionNotFound(5, Default::default(), Default::default()).as_code(); + // partition_id=5, topic_id=1 (numeric), stream_id=2 (numeric) + let mut payload = Vec::new(); + payload.extend_from_slice(&5u64.to_le_bytes()); + payload.push(0); // IdKind::Numeric + payload.push(4); + payload.extend_from_slice(&1u32.to_le_bytes()); + payload.push(0); // IdKind::Numeric + payload.push(4); + payload.extend_from_slice(&2u32.to_le_bytes()); + let body = Bytes::from(payload); + let header = ReplyHeader { + command: Command2::Reply, + size: (HEADER_SIZE + body.len()) as u32, + status: code, + ..Default::default() + }; + let mut buf = [0u8; HEADER_SIZE]; + buf.copy_from_slice(bytemuck::bytes_of(&header)); + let result = decode_response_split(&buf, body); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.as_code(), code); + assert!(matches!(&err, IggyError::PartitionNotFound( + 5, + topic_id, + stream_id + ) if topic_id.value == 1u32.to_le_bytes().to_vec() + && stream_id.value == 2u32.to_le_bytes().to_vec())); + } + + #[test] + fn reply_with_nonzero_data_bearing_code_falls_back_when_body_is_empty() { + let code = + IggyError::PartitionNotFound(5, Default::default(), Default::default()).as_code(); + let header = ReplyHeader { + command: Command2::Reply, + size: HEADER_SIZE as u32, + status: code, + ..Default::default() + }; + let mut buf = [0u8; HEADER_SIZE]; + buf.copy_from_slice(bytemuck::bytes_of(&header)); + let result = decode_response_split(&buf, Bytes::new()); + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.as_code(), code); + assert!(matches!(err, IggyError::PartitionNotFound(..))); + } + #[test] fn reply_with_zero_status_passes_body_through() { // status 0 is the ok channel: a non-metadata reply returns its body. diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index 537d68de22..e1d3464d3f 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -829,6 +829,11 @@ impl WebSocketClient { ); if status != 0 { + let mut body = vec![]; + if length > 0 { + body.resize(length, 0); + stream.read(&mut body).await?; + } // TEMP: See https://github.com/apache/iggy/pull/604 for context. if status == IggyErrorDiscriminants::TopicNameAlreadyExists as u32 || status == IggyErrorDiscriminants::StreamNameAlreadyExists as u32 @@ -849,7 +854,7 @@ impl WebSocketClient { ); } - return Err(IggyError::from_code(status)); + return Err(IggyError::from_code_with_payload(status, &body)); } if length == 0 { diff --git a/core/server/src/sender/mod.rs b/core/server/src/sender/mod.rs index c6a0b4acaa..25b76f0c86 100644 --- a/core/server/src/sender/mod.rs +++ b/core/server/src/sender/mod.rs @@ -203,7 +203,8 @@ pub(crate) async fn send_error_response( where T: AsyncReadExt + AsyncWriteExt + Unpin, { - send_response(stream, &error.as_code().to_le_bytes(), &[]).await + let payload = error.write_payload(); + send_response(stream, &error.as_code().to_le_bytes(), &payload).await } pub(crate) async fn send_response( diff --git a/foreign/python/tests/test_message_operations.py b/foreign/python/tests/test_message_operations.py index da57637c05..a346f3d1e0 100644 --- a/foreign/python/tests/test_message_operations.py +++ b/foreign/python/tests/test_message_operations.py @@ -395,8 +395,8 @@ async def test_poll_messages_with_invalid_partition_id_raises( with pytest.raises( RuntimeError, match=( - r"Partition with ID: 0 for topic with ID: 0 " - r"for stream with ID: 0 was not found\." + r"Partition with ID: 1 for topic with ID: \d+ " + r"for stream with ID: \d+ was not found\." ), ): await iggy_client.poll_messages(