This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 929059e83 refactor!: remove the dead flush unsaved buffer command
(#4317)
929059e83 is described below
commit 929059e83820302c180cd4269d547d610d6c3069
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Mon Sep 28 22:54:36 2026 +0200
refactor!: remove the dead flush unsaved buffer command (#4317)
The server has no on-demand flush. It answers
FLUSH_UNSAVED_BUFFER (code 102) with FeatureUnavailable and
serves no HTTP route for it, yet the Rust SDK, the CLI and the
C#, Node and C++ SDKs still offer a call that always fails.
Remove the command from the protocol table and every client.
The server now refuses 102 like any unknown code, so no special
arm is needed. The C#, Node and Go drift tests pin their code
tables to codes.rs, so all SDKs change together. codes.rs keeps
a note that 102 must never be reused, because older clients
still send it.
---
core/binary_protocol/src/codes.rs | 15 +-
core/binary_protocol/src/dispatch.rs | 203 ++++------------
.../src/requests/messages/flush_unsaved_buffer.rs | 155 ------------
core/binary_protocol/src/requests/messages/mod.rs | 2 -
core/cli/src/args/message.rs | 39 ---
.../src/commands/binary_message/flush_messages.rs | 90 -------
core/cli/src/commands/binary_message/mod.rs | 1 -
core/cli/src/main.rs | 11 +-
.../src/http/messages/flush_unsaved_buffer.rs | 24 --
core/common/src/http/messages/mod.rs | 1 -
core/common/src/traits/binary_impls/messages.rs | 23 +-
core/common/src/traits/message_client.rs | 10 -
core/integration/tests/cli/message/mod.rs | 1 -
.../cli/message/test_message_flush_command.rs | 269 ---------------------
.../tests/cli/message/test_message_help_command.rs | 7 +-
core/integration/tests/data_integrity/mod.rs | 10 +-
.../verify_cluster_replica_data_identical.rs | 2 +-
core/integration/tests/server/flush_vsr.rs | 69 ------
core/integration/tests/server/mod.rs | 2 -
.../server/scenarios/authentication_scenario.rs | 5 -
.../tests/server/scenarios/encryption_scenario.rs | 10 +-
.../scenarios/restart_offset_skip_scenario.rs | 5 +-
.../src/client_wrappers/binary_message_client.rs | 36 ---
core/sdk/src/clients/binary_message.rs | 14 --
core/sdk/src/clients/client.rs | 2 +-
core/sdk/src/http/messages.rs | 34 ---
core/server/src/dispatch/authz.rs | 16 +-
core/server/src/responses.rs | 27 +--
foreign/cpp/src/client.rs | 25 --
foreign/cpp/src/lib.rs | 7 -
foreign/cpp/tests/e2e/client.cpp | 176 --------------
.../FlushMessagesTests.cs | 40 ---
.../Iggy_SDK/Contracts/Auth/GlobalPermissions.cs | 3 -
.../csharp/Iggy_SDK/Contracts/StreamPermissions.cs | 3 -
.../csharp/Iggy_SDK/Contracts/TopicPermissions.cs | 3 -
.../csharp/Iggy_SDK/IggyClient/IIggyPublisher.cs | 17 --
.../Implementations/HttpMessageStream.cs | 10 -
.../IggyClient/Implementations/TcpMessageStream.cs | 10 -
foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs | 1 -
foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs | 4 +-
foreign/csharp/README.md | 3 +-
foreign/go/internal/vsr/protocol_parity_test.go | 7 +-
foreign/node/src/wire/command-set.ts | 4 +-
foreign/node/src/wire/command.code.ts | 1 -
.../wire/message/flush-unsaved-buffers.command.ts | 53 ----
foreign/node/src/wire/message/message.utils.ts | 30 +--
46 files changed, 95 insertions(+), 1385 deletions(-)
diff --git a/core/binary_protocol/src/codes.rs
b/core/binary_protocol/src/codes.rs
index 3a21e1f6a..914edb730 100644
--- a/core/binary_protocol/src/codes.rs
+++ b/core/binary_protocol/src/codes.rs
@@ -53,7 +53,7 @@ pub const LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE: u32 = 44;
// -- Messages --
pub const POLL_MESSAGES_CODE: u32 = 100;
pub const SEND_MESSAGES_CODE: u32 = 101;
-pub const FLUSH_UNSAVED_BUFFER_CODE: u32 = 102;
+// 102 was FLUSH_UNSAVED_BUFFER. Never reuse it: older clients still send it.
pub const GET_POLL_ROUTING_CODE: u32 = 103;
pub const POLL_MESSAGES_ON_PRIMARY_CODE: u32 = 104;
@@ -112,6 +112,8 @@ pub const fn command_name(code: u32) -> Result<&'static
str, WireError> {
mod tests {
use super::*;
+ const RETIRED_CODES: &[u32] = &[102];
+
const ALL_CODES: &[u32] = &[
PING_CODE,
GET_STATS_CODE,
@@ -141,7 +143,6 @@ mod tests {
GET_POLL_ROUTING_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
SEND_MESSAGES_CODE,
- FLUSH_UNSAVED_BUFFER_CODE,
GET_CONSUMER_OFFSET_CODE,
STORE_CONSUMER_OFFSET_CODE,
DELETE_CONSUMER_OFFSET_CODE,
@@ -197,4 +198,14 @@ mod tests {
fn unknown_code_returns_error() {
assert!(command_name(9999).is_err());
}
+
+ #[test]
+ fn retired_codes_remain_unknown() {
+ for &code in RETIRED_CODES {
+ assert!(
+ command_name(code).is_err(),
+ "retired command code {code} must not be reused"
+ );
+ }
+ }
}
diff --git a/core/binary_protocol/src/dispatch.rs
b/core/binary_protocol/src/dispatch.rs
index ad9e2905c..fcb8414bd 100644
--- a/core/binary_protocol/src/dispatch.rs
+++ b/core/binary_protocol/src/dispatch.rs
@@ -113,7 +113,6 @@ pub const COMMAND_TABLE: &[CommandMeta] = &[
// Messages
CommandMeta::non_replicated(POLL_MESSAGES_CODE, "message.poll"),
CommandMeta::replicated(SEND_MESSAGES_CODE, "message.send",
Operation::SendMessages),
- CommandMeta::non_replicated(FLUSH_UNSAVED_BUFFER_CODE,
"message.flush_unsaved_buffer"),
// Consumer Offsets
CommandMeta::non_replicated(GET_CONSUMER_OFFSET_CODE,
"consumer_offset.get"),
CommandMeta::replicated(
@@ -223,38 +222,37 @@ pub const fn lookup_command(code: u32) -> Option<&'static
CommandMeta> {
LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE => 20,
POLL_MESSAGES_CODE => 21,
SEND_MESSAGES_CODE => 22,
- FLUSH_UNSAVED_BUFFER_CODE => 23,
- GET_CONSUMER_OFFSET_CODE => 24,
- STORE_CONSUMER_OFFSET_CODE => 25,
- DELETE_CONSUMER_OFFSET_CODE => 26,
- GET_STREAM_CODE => 27,
- GET_STREAMS_CODE => 28,
- CREATE_STREAM_CODE => 29,
- DELETE_STREAM_CODE => 30,
- UPDATE_STREAM_CODE => 31,
- PURGE_STREAM_CODE => 32,
- GET_TOPIC_CODE => 33,
- GET_TOPICS_CODE => 34,
- CREATE_TOPIC_CODE => 35,
- DELETE_TOPIC_CODE => 36,
- UPDATE_TOPIC_CODE => 37,
- PURGE_TOPIC_CODE => 38,
- CREATE_PARTITIONS_CODE => 39,
- DELETE_PARTITIONS_CODE => 40,
- DELETE_SEGMENTS_CODE => 41,
- GET_CONSUMER_GROUP_CODE => 42,
- GET_CONSUMER_GROUPS_CODE => 43,
- CREATE_CONSUMER_GROUP_CODE => 44,
- DELETE_CONSUMER_GROUP_CODE => 45,
- JOIN_CONSUMER_GROUP_CODE => 46,
- LEAVE_CONSUMER_GROUP_CODE => 47,
- SYNC_CONSUMER_GROUP_CODE => 48,
- LOGIN_REGISTER_WITH_PAT_CODE => 49,
- DESCRIBE_OPTIONS_CODE => 50,
- ATTACH_CONSUMER_SESSION_CODE => 51,
- GET_POLL_ROUTING_CODE => 52,
- POLL_MESSAGES_ON_PRIMARY_CODE => 53,
- GET_CONSUMER_OFFSET_ROUTING_CODE => 54,
+ GET_CONSUMER_OFFSET_CODE => 23,
+ STORE_CONSUMER_OFFSET_CODE => 24,
+ DELETE_CONSUMER_OFFSET_CODE => 25,
+ GET_STREAM_CODE => 26,
+ GET_STREAMS_CODE => 27,
+ CREATE_STREAM_CODE => 28,
+ DELETE_STREAM_CODE => 29,
+ UPDATE_STREAM_CODE => 30,
+ PURGE_STREAM_CODE => 31,
+ GET_TOPIC_CODE => 32,
+ GET_TOPICS_CODE => 33,
+ CREATE_TOPIC_CODE => 34,
+ DELETE_TOPIC_CODE => 35,
+ UPDATE_TOPIC_CODE => 36,
+ PURGE_TOPIC_CODE => 37,
+ CREATE_PARTITIONS_CODE => 38,
+ DELETE_PARTITIONS_CODE => 39,
+ DELETE_SEGMENTS_CODE => 40,
+ GET_CONSUMER_GROUP_CODE => 41,
+ GET_CONSUMER_GROUPS_CODE => 42,
+ CREATE_CONSUMER_GROUP_CODE => 43,
+ DELETE_CONSUMER_GROUP_CODE => 44,
+ JOIN_CONSUMER_GROUP_CODE => 45,
+ LEAVE_CONSUMER_GROUP_CODE => 46,
+ SYNC_CONSUMER_GROUP_CODE => 47,
+ LOGIN_REGISTER_WITH_PAT_CODE => 48,
+ DESCRIBE_OPTIONS_CODE => 49,
+ ATTACH_CONSUMER_SESSION_CODE => 50,
+ GET_POLL_ROUTING_CODE => 51,
+ POLL_MESSAGES_ON_PRIMARY_CODE => 52,
+ GET_CONSUMER_OFFSET_ROUTING_CODE => 53,
_ => return None,
};
Some(&COMMAND_TABLE[idx])
@@ -268,21 +266,21 @@ pub const fn lookup_command(code: u32) -> Option<&'static
CommandMeta> {
pub const fn lookup_by_operation(op: Operation) -> Option<&'static
CommandMeta> {
// Indices must match the order of entries in COMMAND_TABLE above.
let idx = match op {
- Operation::CreateStream => 29,
- Operation::UpdateStream => 31,
- Operation::DeleteStream => 30,
- Operation::PurgeStream => 32,
- Operation::CreateTopic => 35,
- Operation::UpdateTopic => 37,
- Operation::DeleteTopic => 36,
- Operation::PurgeTopic => 38,
- Operation::CreatePartitions => 39,
- Operation::DeletePartitions => 40,
- Operation::DeleteSegments => 41,
- Operation::CreateConsumerGroup => 44,
- Operation::DeleteConsumerGroup => 45,
- Operation::JoinConsumerGroup => 46,
- Operation::LeaveConsumerGroup => 47,
+ Operation::CreateStream => 28,
+ Operation::UpdateStream => 30,
+ Operation::DeleteStream => 29,
+ Operation::PurgeStream => 31,
+ Operation::CreateTopic => 34,
+ Operation::UpdateTopic => 36,
+ Operation::DeleteTopic => 35,
+ Operation::PurgeTopic => 37,
+ Operation::CreatePartitions => 38,
+ Operation::DeletePartitions => 39,
+ Operation::DeleteSegments => 40,
+ Operation::CreateConsumerGroup => 43,
+ Operation::DeleteConsumerGroup => 44,
+ Operation::JoinConsumerGroup => 45,
+ Operation::LeaveConsumerGroup => 46,
Operation::CreateUser => 9,
Operation::UpdateUser => 11,
Operation::DeleteUser => 10,
@@ -291,8 +289,8 @@ pub const fn lookup_by_operation(op: Operation) ->
Option<&'static CommandMeta>
Operation::CreatePersonalAccessToken => 18,
Operation::DeletePersonalAccessToken => 19,
Operation::SendMessages => 22,
- Operation::StoreConsumerOffset => 25,
- Operation::DeleteConsumerOffset => 26,
+ Operation::StoreConsumerOffset => 24,
+ Operation::DeleteConsumerOffset => 25,
Operation::CreateTopicWithAssignments
| Operation::CreatePartitionsWithAssignments
| Operation::RemoveConsumerGroupMember
@@ -310,68 +308,6 @@ pub const fn lookup_by_operation(op: Operation) ->
Option<&'static CommandMeta>
mod tests {
use super::*;
- #[test]
- fn every_code_has_entry() {
- let all_codes = [
- PING_CODE,
- GET_STATS_CODE,
- GET_SNAPSHOT_FILE_CODE,
- GET_CLUSTER_METADATA_CODE,
- GET_ME_CODE,
- GET_CLIENT_CODE,
- GET_CLIENTS_CODE,
- GET_USER_CODE,
- GET_USERS_CODE,
- CREATE_USER_CODE,
- DELETE_USER_CODE,
- UPDATE_USER_CODE,
- UPDATE_PERMISSIONS_CODE,
- CHANGE_PASSWORD_CODE,
- LOGIN_USER_CODE,
- LOGOUT_USER_CODE,
- LOGIN_REGISTER_CODE,
- LOGIN_REGISTER_WITH_PAT_CODE,
- GET_PERSONAL_ACCESS_TOKENS_CODE,
- CREATE_PERSONAL_ACCESS_TOKEN_CODE,
- DELETE_PERSONAL_ACCESS_TOKEN_CODE,
- LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
- POLL_MESSAGES_CODE,
- SEND_MESSAGES_CODE,
- FLUSH_UNSAVED_BUFFER_CODE,
- GET_CONSUMER_OFFSET_CODE,
- STORE_CONSUMER_OFFSET_CODE,
- DELETE_CONSUMER_OFFSET_CODE,
- GET_STREAM_CODE,
- GET_STREAMS_CODE,
- CREATE_STREAM_CODE,
- DELETE_STREAM_CODE,
- UPDATE_STREAM_CODE,
- PURGE_STREAM_CODE,
- GET_TOPIC_CODE,
- GET_TOPICS_CODE,
- CREATE_TOPIC_CODE,
- DELETE_TOPIC_CODE,
- UPDATE_TOPIC_CODE,
- PURGE_TOPIC_CODE,
- CREATE_PARTITIONS_CODE,
- DELETE_PARTITIONS_CODE,
- DELETE_SEGMENTS_CODE,
- GET_CONSUMER_GROUP_CODE,
- GET_CONSUMER_GROUPS_CODE,
- CREATE_CONSUMER_GROUP_CODE,
- DELETE_CONSUMER_GROUP_CODE,
- JOIN_CONSUMER_GROUP_CODE,
- LEAVE_CONSUMER_GROUP_CODE,
- DESCRIBE_OPTIONS_CODE,
- ];
- for code in all_codes {
- assert!(
- lookup_command(code).is_some(),
- "missing dispatch entry for code {code}"
- );
- }
- }
-
#[test]
fn no_duplicate_codes_in_table() {
let mut seen = std::collections::HashSet::new();
@@ -397,49 +333,6 @@ mod tests {
}
}
- #[test]
- fn lookup_by_operation_roundtrips_with_lookup_command() {
- let replicated_ops = [
- Operation::CreateStream,
- Operation::UpdateStream,
- Operation::DeleteStream,
- Operation::PurgeStream,
- Operation::CreateTopic,
- Operation::UpdateTopic,
- Operation::DeleteTopic,
- Operation::PurgeTopic,
- Operation::CreatePartitions,
- Operation::DeletePartitions,
- Operation::DeleteSegments,
- Operation::CreateConsumerGroup,
- Operation::DeleteConsumerGroup,
- Operation::CreateUser,
- Operation::UpdateUser,
- Operation::DeleteUser,
- Operation::ChangePassword,
- Operation::UpdatePermissions,
- Operation::CreatePersonalAccessToken,
- Operation::DeletePersonalAccessToken,
- Operation::JoinConsumerGroup,
- Operation::LeaveConsumerGroup,
- Operation::SendMessages,
- Operation::StoreConsumerOffset,
- Operation::DeleteConsumerOffset,
- ];
- for op in replicated_ops {
- let meta = lookup_by_operation(op)
- .unwrap_or_else(|| panic!("no dispatch entry for operation
{op:?}"));
-
- let by_code = lookup_command(meta.code)
- .unwrap_or_else(|| panic!("no dispatch entry for code {}",
meta.code));
-
- assert_eq!(
- meta.code, by_code.code,
- "lookup_by_operation and lookup disagree for {op:?}"
- );
- }
- }
-
#[test]
fn reserved_operation_returns_none() {
assert!(lookup_by_operation(Operation::Reserved).is_none());
diff --git a/core/binary_protocol/src/requests/messages/flush_unsaved_buffer.rs
b/core/binary_protocol/src/requests/messages/flush_unsaved_buffer.rs
deleted file mode 100644
index 0c276dbec..000000000
--- a/core/binary_protocol/src/requests/messages/flush_unsaved_buffer.rs
+++ /dev/null
@@ -1,155 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use crate::WireError;
-use crate::WireIdentifier;
-use crate::codec::{WireDecode, WireEncode, read_u8, read_u32_le};
-use bytes::{BufMut, BytesMut};
-
-/// `FlushUnsavedBuffer` request.
-///
-/// Wire format: `[stream_id][topic_id][partition_id:4 LE][fsync:1]`
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct FlushUnsavedBufferRequest {
- pub stream_id: WireIdentifier,
- pub topic_id: WireIdentifier,
- pub partition_id: u32,
- pub fsync: bool,
-}
-
-impl WireEncode for FlushUnsavedBufferRequest {
- fn encoded_size(&self) -> usize {
- self.stream_id.encoded_size() + self.topic_id.encoded_size() + 4 + 1
- }
-
- fn encode(&self, buf: &mut BytesMut) {
- self.stream_id.encode(buf);
- self.topic_id.encode(buf);
- buf.put_u32_le(self.partition_id);
- buf.put_u8(u8::from(self.fsync));
- }
-}
-
-impl WireDecode for FlushUnsavedBufferRequest {
- fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
- let mut pos = 0;
- let (stream_id, n) = WireIdentifier::decode(&buf[pos..])?;
- pos += n;
- let (topic_id, n) = WireIdentifier::decode(&buf[pos..])?;
- pos += n;
- let partition_id = read_u32_le(buf, pos)?;
- pos += 4;
- let fsync = read_u8(buf, pos)? != 0;
- pos += 1;
- Ok((
- Self {
- stream_id,
- topic_id,
- partition_id,
- fsync,
- },
- pos,
- ))
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
-
- #[test]
- fn roundtrip() {
- let req = FlushUnsavedBufferRequest {
- stream_id: WireIdentifier::numeric(1),
- topic_id: WireIdentifier::numeric(2),
- partition_id: 3,
- fsync: true,
- };
- let bytes = req.to_bytes();
- let (decoded, consumed) =
FlushUnsavedBufferRequest::decode(&bytes).unwrap();
- assert_eq!(consumed, bytes.len());
- assert_eq!(decoded, req);
- }
-
- #[test]
- fn roundtrip_named_identifiers() {
- let req = FlushUnsavedBufferRequest {
- stream_id: WireIdentifier::named("stream-1").unwrap(),
- topic_id: WireIdentifier::named("topic-1").unwrap(),
- partition_id: 42,
- fsync: false,
- };
- let bytes = req.to_bytes();
- let (decoded, consumed) =
FlushUnsavedBufferRequest::decode(&bytes).unwrap();
- assert_eq!(consumed, bytes.len());
- assert_eq!(decoded, req);
- }
-
- #[test]
- fn fsync_flag_encoding() {
- let req_true = FlushUnsavedBufferRequest {
- stream_id: WireIdentifier::numeric(1),
- topic_id: WireIdentifier::numeric(1),
- partition_id: 0,
- fsync: true,
- };
- let req_false = FlushUnsavedBufferRequest {
- stream_id: WireIdentifier::numeric(1),
- topic_id: WireIdentifier::numeric(1),
- partition_id: 0,
- fsync: false,
- };
- let bytes_true = req_true.to_bytes();
- let bytes_false = req_false.to_bytes();
- assert_eq!(*bytes_true.last().unwrap(), 1);
- assert_eq!(*bytes_false.last().unwrap(), 0);
- }
-
- #[test]
- fn truncated_returns_error() {
- let req = FlushUnsavedBufferRequest {
- stream_id: WireIdentifier::numeric(1),
- topic_id: WireIdentifier::numeric(2),
- partition_id: 3,
- fsync: true,
- };
- let bytes = req.to_bytes();
- for i in 0..bytes.len() {
- assert!(
- FlushUnsavedBufferRequest::decode(&bytes[..i]).is_err(),
- "expected error for truncation at byte {i}"
- );
- }
- }
-
- #[test]
- fn wire_compat_byte_layout() {
- let req = FlushUnsavedBufferRequest {
- stream_id: WireIdentifier::numeric(1),
- topic_id: WireIdentifier::numeric(2),
- partition_id: 3,
- fsync: true,
- };
- let bytes = req.to_bytes();
- // stream_id: [1, 4, 1, 0, 0, 0] + topic_id: [1, 4, 2, 0, 0, 0]
- // + partition_id: [3, 0, 0, 0] + fsync: [1]
- assert_eq!(
- &bytes[..],
- &[1, 4, 1, 0, 0, 0, 1, 4, 2, 0, 0, 0, 3, 0, 0, 0, 1]
- );
- }
-}
diff --git a/core/binary_protocol/src/requests/messages/mod.rs
b/core/binary_protocol/src/requests/messages/mod.rs
index 76584099d..58112850a 100644
--- a/core/binary_protocol/src/requests/messages/mod.rs
+++ b/core/binary_protocol/src/requests/messages/mod.rs
@@ -15,10 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-pub mod flush_unsaved_buffer;
pub mod poll_messages;
pub mod send_messages;
-pub use flush_unsaved_buffer::FlushUnsavedBufferRequest;
pub use poll_messages::PollMessagesRequest;
pub use send_messages::{RawMessage, SendMessagesEncoder, SendMessagesHeader};
diff --git a/core/cli/src/args/message.rs b/core/cli/src/args/message.rs
index f435fd610..56000b836 100644
--- a/core/cli/src/args/message.rs
+++ b/core/cli/src/args/message.rs
@@ -46,22 +46,6 @@ pub(crate) enum MessageAction {
/// iggy message poll --offset 0 stream topic 1
#[clap(verbatim_doc_comment, visible_alias = "p")]
Poll(PollMessagesArgs),
- /// Legacy message flush command (unsupported by the server)
- ///
- /// Binary transports return FeatureUnavailable. HTTP has no flush route.
- /// For acknowledgements backed by stable storage, create the topic with
- /// --durability persisted.
- ///
- /// Stream ID can be specified as a stream name or ID
- /// Topic ID can be specified as a topic name or ID
- ///
- /// Examples:
- /// iggy message flush 1 2 1
- /// iggy message flush stream 2 1
- /// iggy message flush 1 topic 1
- /// iggy message flush stream topic 1
- #[clap(verbatim_doc_comment, visible_alias = "f")]
- Flush(FlushMessagesArgs),
}
#[derive(Debug, Clone, Args)]
@@ -266,29 +250,6 @@ pub(crate) struct PollMessagesArgs {
pub(crate) output_file: Option<String>,
}
-#[derive(Debug, Clone, Args)]
-pub(crate) struct FlushMessagesArgs {
- /// Stream ID for the flush request
- ///
- /// Stream ID can be specified as a stream name or ID
- #[arg(value_parser = clap::value_parser!(Identifier))]
- pub(crate) stream_id: Identifier,
- /// Topic ID for the flush request
- ///
- /// Topic ID can be specified as a topic name or ID
- #[arg(value_parser = clap::value_parser!(Identifier))]
- pub(crate) topic_id: Identifier,
- /// Partition ID for the flush request
- #[arg(value_parser = clap::value_parser!(u32).range(0..))]
- pub(crate) partition_id: u32,
- /// Request fsync (unsupported by the server)
- ///
- /// The server rejects flush requests regardless of this flag.
- #[clap(verbatim_doc_comment)]
- #[clap(short, long, default_value_t = false)]
- pub(crate) fsync: bool,
-}
-
#[cfg(test)]
mod tests {
use super::*;
diff --git a/core/cli/src/commands/binary_message/flush_messages.rs
b/core/cli/src/commands/binary_message/flush_messages.rs
deleted file mode 100644
index e74041d93..000000000
--- a/core/cli/src/commands/binary_message/flush_messages.rs
+++ /dev/null
@@ -1,90 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use crate::commands::cli_command::{CliCommand, PRINT_TARGET};
-use anyhow::{Context, Error};
-use async_trait::async_trait;
-use iggy_common::Client;
-use iggy_common::Identifier;
-use tracing::{Level, event};
-
-pub struct FlushMessagesCmd {
- stream_id: Identifier,
- topic_id: Identifier,
- partition_id: u32,
- fsync: bool,
-}
-
-impl FlushMessagesCmd {
- pub fn new(
- stream_id: Identifier,
- topic_id: Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Self {
- Self {
- stream_id,
- topic_id,
- partition_id,
- fsync,
- }
- }
-}
-
-#[async_trait]
-impl CliCommand for FlushMessagesCmd {
- fn explain(&self) -> String {
- format!(
- "flush messages from topic with ID: {} and stream with ID: {}
(partition with ID: {}) {}",
- self.topic_id,
- self.stream_id,
- self.partition_id,
- if self.fsync {
- "with fsync"
- } else {
- "without fsync"
- },
- )
- }
-
- async fn execute_cmd(&mut self, client: &dyn Client) -> anyhow::Result<(),
Error> {
- client
- .flush_unsaved_buffer(
- &self.stream_id,
- &self.topic_id,
- self.partition_id,
- self.fsync,
- )
- .await
- .with_context(|| {
- format!(
- "Problem flushing messages from topic with ID: {} and
stream with ID: {} (partition with ID: {}) {}",
- self.topic_id, self.stream_id, self.partition_id, if
self.fsync { "with fsync" } else { "without fsync" },
- )
- })?;
-
- event!(target: PRINT_TARGET, Level::INFO,
- "Flushed messages from topic with ID: {} and stream with ID: {}
(partition with ID: {}) {}",
- self.topic_id,
- self.stream_id,
- self.partition_id,
- if self.fsync { "with fsync" } else { "without fsync" },
- );
-
- Ok(())
- }
-}
diff --git a/core/cli/src/commands/binary_message/mod.rs
b/core/cli/src/commands/binary_message/mod.rs
index 9f61d5c94..ba2da6a3f 100644
--- a/core/cli/src/commands/binary_message/mod.rs
+++ b/core/cli/src/commands/binary_message/mod.rs
@@ -15,6 +15,5 @@
// specific language governing permissions and limitations
// under the License.
-pub mod flush_messages;
pub mod poll_messages;
pub mod send_messages;
diff --git a/core/cli/src/main.rs b/core/cli/src/main.rs
index 7849d390f..ad2909e10 100644
--- a/core/cli/src/main.rs
+++ b/core/cli/src/main.rs
@@ -61,10 +61,7 @@ use iggy_cli::commands::{
get_consumer_offset::GetConsumerOffsetCmd,
set_consumer_offset::SetConsumerOffsetCmd,
},
binary_context::get_contexts::GetContextsCmd,
- binary_message::{
- flush_messages::FlushMessagesCmd, poll_messages::PollMessagesCmd,
- send_messages::SendMessagesCmd,
- },
+ binary_message::{poll_messages::PollMessagesCmd,
send_messages::SendMessagesCmd},
binary_partitions::{
create_partitions::CreatePartitionsCmd,
delete_partitions::DeletePartitionsCmd,
},
@@ -306,12 +303,6 @@ fn get_command(
poll_args.show_headers,
poll_args.output_file.clone(),
)),
- MessageAction::Flush(flush_args) => Box::new(FlushMessagesCmd::new(
- flush_args.stream_id.clone(),
- flush_args.topic_id.clone(),
- flush_args.partition_id,
- flush_args.fsync,
- )),
},
Command::ConsumerOffset(command) => match command {
ConsumerOffsetAction::Get(get_args) =>
Box::new(GetConsumerOffsetCmd::new(
diff --git a/core/common/src/http/messages/flush_unsaved_buffer.rs
b/core/common/src/http/messages/flush_unsaved_buffer.rs
deleted file mode 100644
index 7ba95fe7f..000000000
--- a/core/common/src/http/messages/flush_unsaved_buffer.rs
+++ /dev/null
@@ -1,24 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use serde::{Deserialize, Serialize};
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct FlushUnsavedBuffer {
- pub partition_id: u32,
- pub fsync: bool,
-}
diff --git a/core/common/src/http/messages/mod.rs
b/core/common/src/http/messages/mod.rs
index 105ca06d8..ba2da6a3f 100644
--- a/core/common/src/http/messages/mod.rs
+++ b/core/common/src/http/messages/mod.rs
@@ -15,6 +15,5 @@
// specific language governing permissions and limitations
// under the License.
-pub mod flush_unsaved_buffer;
pub mod poll_messages;
pub mod send_messages;
diff --git a/core/common/src/traits/binary_impls/messages.rs
b/core/common/src/traits/binary_impls/messages.rs
index 2dd6679f9..1c42efe74 100644
--- a/core/common/src/traits/binary_impls/messages.rs
+++ b/core/common/src/traits/binary_impls/messages.rs
@@ -28,11 +28,11 @@ use crate::{ConsumerKind, PartitioningKind, TopicClient,
calculate_32};
use bytes::BytesMut;
use iggy_binary_protocol::codec::WireDecode;
use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::codes::SEND_MESSAGES_CODE;
use iggy_binary_protocol::codes::SYNC_CONSUMER_GROUP_CODE;
-use iggy_binary_protocol::codes::{FLUSH_UNSAVED_BUFFER_CODE,
SEND_MESSAGES_CODE};
use iggy_binary_protocol::requests::consumer_groups::SyncConsumerGroupRequest;
use iggy_binary_protocol::requests::messages::{
- FlushUnsavedBufferRequest, PollMessagesRequest, RawMessage,
SendMessagesEncoder,
+ PollMessagesRequest, RawMessage, SendMessagesEncoder,
};
use
iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse;
@@ -399,25 +399,6 @@ impl<B: BinaryClient> MessageClient for B {
.await?;
Ok(committed_send_confirmations(&response))
}
-
- async fn flush_unsaved_buffer(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Result<(), IggyError> {
- fail_if_not_authenticated(self).await?;
- let req = FlushUnsavedBufferRequest {
- stream_id: identifier_to_wire(stream_id)?,
- topic_id: identifier_to_wire(topic_id)?,
- partition_id,
- fsync,
- };
- self.send_raw_with_response(FLUSH_UNSAVED_BUFFER_CODE, req.to_bytes())
- .await?;
- Ok(())
- }
}
#[cfg(test)]
diff --git a/core/common/src/traits/message_client.rs
b/core/common/src/traits/message_client.rs
index 1bd3ab13e..2ca4db532 100644
--- a/core/common/src/traits/message_client.rs
+++ b/core/common/src/traits/message_client.rs
@@ -134,14 +134,4 @@ pub trait MessageClient {
partitioning: &Partitioning,
messages: &mut [IggyMessage],
) -> Result<SendMessagesResponse, IggyError>;
-
- /// Force flush of the `unsaved_messages` buffer to disk, optionally
fsyncing the data.
- #[allow(clippy::too_many_arguments)]
- async fn flush_unsaved_buffer(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Result<(), IggyError>;
}
diff --git a/core/integration/tests/cli/message/mod.rs
b/core/integration/tests/cli/message/mod.rs
index cbb7fc58d..ce7a0a8da 100644
--- a/core/integration/tests/cli/message/mod.rs
+++ b/core/integration/tests/cli/message/mod.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-mod test_message_flush_command;
mod test_message_help_command;
mod test_message_poll_command;
mod test_message_poll_to_file_command;
diff --git a/core/integration/tests/cli/message/test_message_flush_command.rs
b/core/integration/tests/cli/message/test_message_flush_command.rs
deleted file mode 100644
index d1501243a..000000000
--- a/core/integration/tests/cli/message/test_message_flush_command.rs
+++ /dev/null
@@ -1,269 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-// Tests the `message flush` CLI command itself (`FLUSH_UNSAVED_BUFFER`), not
-// a durability barrier some other assertion needs. The command is slated for
-// removal from the protocol, so this stays out of the vsr pass permanently
-// (today via the whole-`cli`-module gate): delete this file together with the
-// command, do not adapt it.
-use crate::cli::common::{
- CLAP_INDENT, IggyCmdCommand, IggyCmdTest, IggyCmdTestCase, TestHelpCmd,
USAGE_PREFIX,
-};
-use assert_cmd::assert::Assert;
-use async_trait::async_trait;
-use iggy::prelude::Client;
-use iggy::prelude::Identifier;
-use iggy::prelude::IggyExpiry;
-use iggy::prelude::TopicCreateOptions;
-use predicates::str::contains;
-use serial_test::parallel;
-use std::str::FromStr;
-
-struct TestMessageFetchCmd {
- stream_name: String,
- topic_name: String,
- partitions_count: u32,
- partition_id: u32,
- fsync: bool,
- // These will be populated after creating the resources
- actual_stream_id: Option<u32>,
- actual_topic_id: Option<u32>,
-}
-
-impl TestMessageFetchCmd {
- fn new(
- stream_name: &str,
- topic_name: &str,
- partitions_count: u32,
- partition_id: u32,
- fsync: bool,
- ) -> Self {
- Self {
- stream_name: stream_name.to_string(),
- topic_name: topic_name.to_string(),
- partitions_count,
- partition_id,
- fsync,
- actual_stream_id: None,
- actual_topic_id: None,
- }
- }
-
- fn to_args(&self) -> Vec<String> {
- let mut command = Vec::new();
-
- // Use actual stream ID if available, otherwise use stream name as
fallback
- if let Some(stream_id) = self.actual_stream_id {
- command.push(format!("{}", stream_id));
- } else {
- command.push(self.stream_name.clone());
- }
-
- // Use actual topic ID if available, otherwise use topic name as
fallback
- if let Some(topic_id) = self.actual_topic_id {
- command.push(format!("{}", topic_id));
- } else {
- command.push(self.topic_name.clone());
- }
-
- command.push(format!("{}", self.partition_id));
-
- if self.fsync {
- command.push("--fsync".to_string());
- }
-
- command
- }
-}
-
-#[async_trait]
-impl IggyCmdTestCase for TestMessageFetchCmd {
- async fn prepare_server_state(&mut self, client: &dyn Client) {
- // Create stream and capture its actual ID
- let stream = client
- .create_stream(&self.stream_name)
- .await
- .expect("Failed to create stream");
- self.actual_stream_id = Some(stream.id);
-
- // Create topic and capture its actual ID
- let topic = client
- .create_topic(
- &stream.id.try_into().unwrap(),
- &self.topic_name,
- &TopicCreateOptions {
- partitions_count: Some(self.partitions_count),
- message_expiry: Some(IggyExpiry::NeverExpire),
- ..TopicCreateOptions::default()
- },
- )
- .await
- .expect("Failed to create topic");
- self.actual_topic_id = Some(topic.id);
- }
-
- fn get_command(&self) -> IggyCmdCommand {
- IggyCmdCommand::new()
- .arg("message")
- .arg("flush")
- .args(self.to_args())
- .with_env_credentials()
- }
-
- fn verify_command(&self, command_state: Assert) {
- let stream_id = if let Some(stream_id) = self.actual_stream_id {
- format!("{}", stream_id)
- } else {
- self.stream_name.clone()
- };
-
- let topic_id = if let Some(topic_id) = self.actual_topic_id {
- format!("{}", topic_id)
- } else {
- self.topic_name.clone()
- };
-
- let identification_part = format!(
- "from topic with ID: {topic_id} and stream with ID: {stream_id}
(partition with ID: {}) {}",
- self.partition_id,
- if self.fsync {
- "with fsync"
- } else {
- "without fsync"
- }
- );
-
- // The server has no on-demand flush primitive: FLUSH_UNSAVED_BUFFER
- // surfaces a typed FeatureUnavailable (see flush_vsr.rs), so the CLI
- // reports a flush problem instead of success.
- command_state.failure().stderr(contains(format!(
- "Problem flushing messages {identification_part}"
- )));
- }
-
- async fn verify_server_state(&self, client: &dyn Client) {
- if let Some(topic_id) = self.actual_topic_id {
- let stream_id = Identifier::from_str(&self.stream_name);
- assert!(stream_id.is_ok());
- let stream_id = stream_id.unwrap();
-
- let topic = client
- .delete_topic(&stream_id, &topic_id.try_into().unwrap())
- .await;
- assert!(topic.is_ok());
-
- let stream = client.delete_stream(&stream_id).await;
- assert!(stream.is_ok());
- }
- }
-}
-
-#[tokio::test]
-#[parallel]
-pub async fn should_be_successful() {
- let mut iggy_cmd_test = IggyCmdTest::default();
-
- let test_parameters = vec![false, true]; // Test with and without fsync
-
- iggy_cmd_test.setup().await;
- for fsync in test_parameters {
- iggy_cmd_test
- .execute_test(TestMessageFetchCmd::new("stream", "topic", 1, 0,
fsync))
- .await;
- }
-}
-
-#[tokio::test]
-#[parallel]
-pub async fn should_help_match() {
- let mut iggy_cmd_test = IggyCmdTest::help_message();
-
- iggy_cmd_test
- .execute_test_for_help_command(TestHelpCmd::new(
- vec!["message", "flush", "--help"],
- format!(
- r#"Legacy message flush command (unsupported by the server)
-
-Binary transports return FeatureUnavailable. HTTP has no flush route.
-For acknowledgements backed by stable storage, create the topic with
---durability persisted.
-
-Stream ID can be specified as a stream name or ID
-Topic ID can be specified as a topic name or ID
-
-Examples:
- iggy message flush 1 2 1
- iggy message flush stream 2 1
- iggy message flush 1 topic 1
- iggy message flush stream topic 1
-
-{USAGE_PREFIX} message flush [OPTIONS] <STREAM_ID> <TOPIC_ID> <PARTITION_ID>
-
-Arguments:
- <STREAM_ID>
- Stream ID for the flush request
-{CLAP_INDENT}
- Stream ID can be specified as a stream name or ID
-
- <TOPIC_ID>
- Topic ID for the flush request
-{CLAP_INDENT}
- Topic ID can be specified as a topic name or ID
-
- <PARTITION_ID>
- Partition ID for the flush request
-
-Options:
- -f, --fsync
- Request fsync (unsupported by the server)
-{CLAP_INDENT}
- The server rejects flush requests regardless of this flag.
-
- -h, --help
- Print help (see a summary with '-h')
-"#,
- ),
- ))
- .await;
-}
-
-#[tokio::test]
-#[parallel]
-pub async fn should_short_help_match() {
- let mut iggy_cmd_test = IggyCmdTest::default();
-
- iggy_cmd_test
- .execute_test_for_help_command(TestHelpCmd::new(
- vec!["message", "flush", "-h"],
- format!(
- r#"Legacy message flush command (unsupported by the server)
-
-{USAGE_PREFIX} message flush [OPTIONS] <STREAM_ID> <TOPIC_ID> <PARTITION_ID>
-
-Arguments:
- <STREAM_ID> Stream ID for the flush request
- <TOPIC_ID> Topic ID for the flush request
- <PARTITION_ID> Partition ID for the flush request
-
-Options:
- -f, --fsync Request fsync (unsupported by the server)
- -h, --help Print help (see more with '--help')
-"#,
- ),
- ))
- .await;
-}
diff --git a/core/integration/tests/cli/message/test_message_help_command.rs
b/core/integration/tests/cli/message/test_message_help_command.rs
index 83e83d010..e1c12cc4c 100644
--- a/core/integration/tests/cli/message/test_message_help_command.rs
+++ b/core/integration/tests/cli/message/test_message_help_command.rs
@@ -32,10 +32,9 @@ pub async fn should_help_match() {
{USAGE_PREFIX} message <COMMAND>
Commands:
- send Send messages to given topic ID and given stream ID [alias: s]
- poll Poll messages from given topic ID and given stream ID [alias: p]
- flush Legacy message flush command (unsupported by the server) [alias: f]
- help Print this message or the help of the given subcommand(s)
+ send Send messages to given topic ID and given stream ID [alias: s]
+ poll Poll messages from given topic ID and given stream ID [alias: p]
+ help Print this message or the help of the given subcommand(s)
Options:
-h, --help Print help
diff --git a/core/integration/tests/data_integrity/mod.rs
b/core/integration/tests/data_integrity/mod.rs
index 3386feccb..fece21969 100644
--- a/core/integration/tests/data_integrity/mod.rs
+++ b/core/integration/tests/data_integrity/mod.rs
@@ -15,12 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-// Partially vsr-gated inside the module: the remaining gates cover
-// `flush_unsaved_buffer`, which the server answers `FeatureUnavailable` and
-// which the eager-flush server envs replace under vsr. The bench-fill test
-// itself runs under vsr since PARTITION-plane state transfer landed, but the
-// harness spawns `iggy-bench` off disk with no cargo build-graph edge, so the
-// binary must be freshly built or a stale build hangs on login.
+// The bench-fill test runs under vsr since PARTITION-plane state transfer
+// landed, but the harness spawns `iggy-bench` off disk with no cargo
+// build-graph edge, so the binary must be freshly built or a stale build
+// hangs on login.
mod verify_after_server_restart;
mod verify_user_login_after_restart;
diff --git
a/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs
b/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs
index 668b5f118..dc8b991b5 100644
---
a/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs
+++
b/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs
@@ -67,7 +67,7 @@ const BURST_BATCH_MESSAGES: u32 = 2;
// `messages_required_to_save = 1` forces every committed batch to persist to
its
// segment immediately on every node, so each replica materialises the segment
-// files while running (the VSR server serves no flush_unsaved_buffer, and
+// files while running (the server has no on-demand flush, and
// shutdown-flush would couple the test to drain behaviour). It is a topic
// creation option, so it travels with the topic to every replica.
#[iggy_harness(cluster_nodes = 3)]
diff --git a/core/integration/tests/server/flush_vsr.rs
b/core/integration/tests/server/flush_vsr.rs
deleted file mode 100644
index adeee0dd7..000000000
--- a/core/integration/tests/server/flush_vsr.rs
+++ /dev/null
@@ -1,69 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-//! Flush contract against the server (vsr): the server has no on-demand flush
-//! primitive, so `FLUSH_UNSAVED_BUFFER` must surface a typed
-//! `FeatureUnavailable` over the SDK rather than the non-replicated
catch-all's
-//! empty-ok, which would fake a durability guarantee.
-//!
-//! `FLUSH_UNSAVED_BUFFER` is slated for removal from the protocol. This test
-//! pins the deny contract only until then: delete this file together with the
-//! command (and the SDK method), do not port it anywhere.
-
-use iggy::prelude::*;
-use integration::iggy_harness;
-
-// Partition ids are 0-based (CreateTopic assigns them from 0).
-const PARTITION_ID: u32 = 0;
-
-#[iggy_harness(
- test_client_transport = [Tcp]
-)]
-async fn given_valid_partition_when_flushing_should_reject_feature_unavailable(
- harness: &TestHarness,
-) {
- let client = harness.tcp_root_client().await.expect("tcp root client");
- client
- .create_stream("flush-stream")
- .await
- .expect("create stream");
- let stream_id = Identifier::from_str_value("flush-stream").expect("stream
identifier");
- client
- .create_topic(
- &stream_id,
- "flush-topic",
- &TopicCreateOptions {
- partitions_count: Some(1),
- message_expiry: Some(IggyExpiry::NeverExpire),
- ..TopicCreateOptions::default()
- },
- )
- .await
- .expect("create topic");
- let topic_id = Identifier::from_str_value("flush-topic").expect("topic
identifier");
-
- // A valid, existing partition proves the deny is feature-unavailable, not
a
- // target-not-found in disguise; fsync makes it the strongest durability
ask.
- let result = client
- .flush_unsaved_buffer(&stream_id, &topic_id, PARTITION_ID, true)
- .await;
-
- assert!(
- matches!(&result, Err(error) if error.as_code() ==
IggyError::FeatureUnavailable.as_code()),
- "flush over TCP-ng must surface Err(FeatureUnavailable), got
{result:?}"
- );
-}
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index 159b9217b..7da753a51 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -19,8 +19,6 @@
// the server's ported trusted-issuer path verify them.
mod a2a_jwt;
mod cg;
-// Flush (FLUSH_UNSAVED_BUFFER) has no the server primitive; it must deny
typed.
-mod flush_vsr;
// Raw TCP framing (connect, hand-crafted frames, root register) for the
// server suites that send what the SDK cannot.
pub(crate) mod raw_tcp;
diff --git a/core/integration/tests/server/scenarios/authentication_scenario.rs
b/core/integration/tests/server/scenarios/authentication_scenario.rs
index 19473c7d0..5f9bf7e28 100644
--- a/core/integration/tests/server/scenarios/authentication_scenario.rs
+++ b/core/integration/tests/server/scenarios/authentication_scenario.rs
@@ -305,11 +305,6 @@ async fn test_all_commands_require_auth(client:
&IggyClient) {
)
.await
.map(|_| ()),
- FLUSH_UNSAVED_BUFFER_CODE => {
- client
- .flush_unsaved_buffer(&ctx.stream_id, &ctx.topic_id, 0,
false)
- .await
- }
// Consumer Offsets
GET_CONSUMER_OFFSET_ROUTING_CODE => client
diff --git a/core/integration/tests/server/scenarios/encryption_scenario.rs
b/core/integration/tests/server/scenarios/encryption_scenario.rs
index a59c515ef..8bab9d59d 100644
--- a/core/integration/tests/server/scenarios/encryption_scenario.rs
+++ b/core/integration/tests/server/scenarios/encryption_scenario.rs
@@ -97,9 +97,8 @@ async fn
should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
- // No flush primitive exists (FLUSH_UNSAVED_BUFFER denies typed); the
- // eager-flush envs in `build_server_config` make every committed batch hit
- // disk instead.
+ // No flush primitive exists; `eager_flush_options()` makes every committed
+ // batch hit disk instead.
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
// Verify on-disk encryption of headers and payload
@@ -244,9 +243,8 @@ async fn
should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
- // No flush primitive exists (FLUSH_UNSAVED_BUFFER denies typed); the
- // eager-flush envs in `build_server_config` make every committed batch hit
- // disk instead.
+ // No flush primitive exists; `eager_flush_options()` makes every committed
+ // batch hit disk instead.
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
let polled = client
diff --git
a/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
b/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
index c97b42582..d4eb19fff 100644
--- a/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
+++ b/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
@@ -99,9 +99,8 @@ pub async fn run(harness: &mut TestHarness) {
.await
.unwrap();
- // No flush primitive exists (FLUSH_UNSAVED_BUFFER denies typed); graceful
- // shutdown flushes the committed journal, which this scenario's restart
- // exercises instead.
+ // No flush primitive exists; graceful shutdown flushes the committed
+ // journal, which this scenario's restart exercises instead.
tokio::time::sleep(Duration::from_secs(2)).await;
drop(setup_client);
diff --git a/core/sdk/src/client_wrappers/binary_message_client.rs
b/core/sdk/src/client_wrappers/binary_message_client.rs
index 3fb72563a..1d19ffba8 100644
--- a/core/sdk/src/client_wrappers/binary_message_client.rs
+++ b/core/sdk/src/client_wrappers/binary_message_client.rs
@@ -161,40 +161,4 @@ impl MessageClient for ClientWrapper {
}
}
}
-
- async fn flush_unsaved_buffer(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partitioning_id: u32,
- fsync: bool,
- ) -> Result<(), IggyError> {
- match self {
- ClientWrapper::Iggy(client) => {
- client
- .flush_unsaved_buffer(stream_id, topic_id,
partitioning_id, fsync)
- .await
- }
- ClientWrapper::Http(client) => {
- client
- .flush_unsaved_buffer(stream_id, topic_id,
partitioning_id, fsync)
- .await
- }
- ClientWrapper::Tcp(client) => {
- client
- .flush_unsaved_buffer(stream_id, topic_id,
partitioning_id, fsync)
- .await
- }
- ClientWrapper::Quic(client) => {
- client
- .flush_unsaved_buffer(stream_id, topic_id,
partitioning_id, fsync)
- .await
- }
- ClientWrapper::WebSocket(client) => {
- client
- .flush_unsaved_buffer(stream_id, topic_id,
partitioning_id, fsync)
- .await
- }
- }
- }
}
diff --git a/core/sdk/src/clients/binary_message.rs
b/core/sdk/src/clients/binary_message.rs
index e83321bbd..7e2e1f7b1 100644
--- a/core/sdk/src/clients/binary_message.rs
+++ b/core/sdk/src/clients/binary_message.rs
@@ -125,18 +125,4 @@ impl MessageClient for IggyClient {
.send_messages(stream_id, topic_id, partitioning, messages)
.await
}
-
- async fn flush_unsaved_buffer(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Result<(), IggyError> {
- self.client
- .read()
- .await
- .flush_unsaved_buffer(stream_id, topic_id, partition_id, fsync)
- .await
- }
}
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 09b8beff4..2f646ad06 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -100,7 +100,7 @@ const SESSION_CONTROL_CODES: [u32; 5] = [
/// - [`SegmentClient`]: delete closed segments from a partition.
/// - [`ConsumerGroupClient`]: create, get, delete, and join or leave consumer
groups.
/// - [`ConsumerOffsetClient`]: store, read, and delete consumer offsets.
-/// - [`MessageClient`]: send and poll messages, and flush the unsaved buffer.
+/// - [`MessageClient`]: send and poll messages.
///
/// Additionally, you can bypass invoking methods from these traits and
directly talk to the server with [`send_binary_request`] and
[`send_http_request`] for http.
/// Both trade typed API's safety for low-level control. You need to know the
server codes and the wire format.
diff --git a/core/sdk/src/http/messages.rs b/core/sdk/src/http/messages.rs
index d12f76b24..7a18f03e1 100644
--- a/core/sdk/src/http/messages.rs
+++ b/core/sdk/src/http/messages.rs
@@ -25,7 +25,6 @@ use async_trait::async_trait;
use iggy_common::IggyMessagesBatch;
use iggy_common::MessageClient;
use iggy_common::SendMessagesConfirmations;
-use iggy_common::flush_unsaved_buffer::FlushUnsavedBuffer;
#[async_trait]
impl MessageClient for HttpClient {
@@ -72,30 +71,6 @@ impl MessageClient for HttpClient {
.await?;
decode_send_response(response).await
}
-
- async fn flush_unsaved_buffer(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Result<(), IggyError> {
- let _ = self
- .get_with_query(
- &get_path_flush_unsaved_buffer(
- &stream_id.as_cow_str(),
- &topic_id.as_cow_str(),
- partition_id,
- fsync,
- ),
- &FlushUnsavedBuffer {
- partition_id,
- fsync,
- },
- )
- .await?;
- Ok(())
- }
}
impl HttpClient {
@@ -172,15 +147,6 @@ fn get_path(stream_id: &str, topic_id: &str) -> String {
format!("streams/{stream_id}/topics/{topic_id}/messages")
}
-fn get_path_flush_unsaved_buffer(
- stream_id: &str,
- topic_id: &str,
- partition_id: u32,
- fsync: bool,
-) -> String {
-
format!("streams/{stream_id}/topics/{topic_id}/messages/flush/{partition_id}/fsync={fsync}")
-}
-
#[cfg(test)]
mod tests {
use super::{SendMessagesConfirmations, SendMessagesResponse};
diff --git a/core/server/src/dispatch/authz.rs
b/core/server/src/dispatch/authz.rs
index 9151a8dff..a00f65bd4 100644
--- a/core/server/src/dispatch/authz.rs
+++ b/core/server/src/dispatch/authz.rs
@@ -29,10 +29,9 @@ use std::rc::Rc;
use consensus::MetadataHandle;
use iggy_binary_protocol::codes::{
- DESCRIBE_OPTIONS_CODE, FLUSH_UNSAVED_BUFFER_CODE,
GET_CLUSTER_METADATA_CODE,
- GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE,
GET_PERSONAL_ACCESS_TOKENS_CODE,
- GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE,
GET_TOPICS_CODE,
- GET_USER_CODE, GET_USERS_CODE,
+ DESCRIBE_OPTIONS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE,
+ GET_CONSUMER_GROUPS_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE,
GET_STREAM_CODE,
+ GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE,
GET_USERS_CODE,
};
use iggy_binary_protocol::requests::consumer_groups::{
GetConsumerGroupRequest, GetConsumerGroupsRequest,
@@ -265,9 +264,6 @@ where
|request| (&request.stream_id, &request.topic_id),
Permissioner::get_consumer_groups,
),
- // No on-demand flush primitive exists, and flush has no HTTP route, so
- // this arm is the only thing answering `FeatureUnavailable` for it.
- FLUSH_UNSAVED_BUFFER_CODE => Err(IggyError::FeatureUnavailable),
// A replicated code smuggled inside a `NonReplicated` header keeps the
// builder's `FeatureUnavailable`; a table-listed code with no arm
above
// and an unknown code are both refused as `InvalidCommand`. The
builder
@@ -446,7 +442,6 @@ mod tests {
let allow = Verdict::Allow;
let unauthenticated =
Verdict::Deny(IggyError::Unauthenticated.as_code());
let invalid_command =
Verdict::Deny(IggyError::InvalidCommand.as_code());
- let feature_unavailable =
Verdict::Deny(IggyError::FeatureUnavailable.as_code());
vec![
(PING_CODE, invalid_command, invalid_command),
(GET_STATS_CODE, unauthenticated, allow),
@@ -483,11 +478,6 @@ mod tests {
invalid_command,
invalid_command,
),
- (
- FLUSH_UNSAVED_BUFFER_CODE,
- feature_unavailable,
- feature_unavailable,
- ),
(GET_CONSUMER_OFFSET_CODE, invalid_command, invalid_command),
(GET_STREAM_CODE, allow, allow),
(GET_STREAMS_CODE, unauthenticated, allow),
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 05e1995be..2773b07a6 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -33,10 +33,9 @@ use bytes::Bytes;
use consensus::MetadataHandle;
use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::codes::{
- DESCRIBE_OPTIONS_CODE, FLUSH_UNSAVED_BUFFER_CODE,
GET_CLUSTER_METADATA_CODE,
- GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE,
GET_PERSONAL_ACCESS_TOKENS_CODE,
- GET_SNAPSHOT_FILE_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE,
GET_TOPIC_CODE,
- GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
+ DESCRIBE_OPTIONS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE,
+ GET_CONSUMER_GROUPS_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE,
GET_STREAM_CODE,
+ GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE,
GET_USERS_CODE,
};
use iggy_binary_protocol::requests::consumer_groups::{
GetConsumerGroupRequest, GetConsumerGroupsRequest,
@@ -332,22 +331,10 @@ where
}
GET_CONSUMER_GROUP_CODE => build_consumer_group_response(shard, body),
GET_CONSUMER_GROUPS_CODE => build_consumer_groups_response(shard,
body),
- // The server has no on-demand flush primitive, so it denies honestly.
- // The non-replicated catch-all's empty-ok would otherwise attest a
- // durability guarantee the server never gave.
- FLUSH_UNSAVED_BUFFER_CODE => Err(IggyError::FeatureUnavailable),
- // Snapshot collection blocks on shell-outs, so the dedicated dispatch
- // and HTTP handlers await it off-thread; this synchronous builder
- // cannot, and reaching it here is a routing bug. Fail closed rather
- // than let the catch-all's empty-ok attest an artifact that was never
- // produced.
- GET_SNAPSHOT_FILE_CODE => Err(IggyError::InvalidCommand),
- // Sequenced AFTER the named arms above, so flush keeps answering
- // `FeatureUnavailable`. A table-listed non-replicated code with no arm
- // is a routing bug and an unknown code is a client bug; the empty-ok
- // that used to cover both attested a read that never ran. Only the
- // named arms return `Empty`, and there it means "resolved to nothing"
- // (the 404 the HTTP path maps).
+ // A table-listed non-replicated code with no arm is a routing bug and
+ // an unknown code is a client bug; the empty-ok that used to cover
both
+ // attested a read that never ran. Only the named arms return `Empty`,
+ // and there it means "resolved to nothing" (the 404 the HTTP path
maps).
_ => match iggy_binary_protocol::dispatch::lookup_command(code) {
Some(meta) if meta.is_replicated() =>
Err(IggyError::FeatureUnavailable),
_ => Err(IggyError::InvalidCommand),
diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs
index 6ea8fdc9e..9cd8e3d55 100644
--- a/foreign/cpp/src/client.rs
+++ b/foreign/cpp/src/client.rs
@@ -359,31 +359,6 @@ impl Client {
})
}
- pub fn flush_unsaved_buffer(
- &self,
- stream_id: ffi::Identifier,
- topic_id: ffi::Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Result<(), String> {
- let rust_stream_id = RustIdentifier::try_from(stream_id)
- .map_err(|error| format!("Could not flush unsaved buffer:
{error}"))?;
- let rust_topic_id = RustIdentifier::try_from(topic_id)
- .map_err(|error| format!("Could not flush unsaved buffer:
{error}"))?;
-
- RUNTIME.block_on(async {
- self.inner
- .flush_unsaved_buffer(&rust_stream_id, &rust_topic_id,
partition_id, fsync)
- .await
- .map_err(|error| {
- format!(
- "Could not flush unsaved buffer for stream
'{rust_stream_id}', topic '{rust_topic_id}', partition '{partition_id}':
{error}"
- )
- })?;
- Ok(())
- })
- }
-
#[allow(clippy::too_many_arguments)]
pub fn poll_messages(
&self,
diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs
index 3ed6d1103..b07c4b93b 100644
--- a/foreign/cpp/src/lib.rs
+++ b/foreign/cpp/src/lib.rs
@@ -604,13 +604,6 @@ mod ffi {
partitioning_value: Vec<u8>,
messages: Vec<IggyMessageToSend>,
) -> Result<SendMessagesResponse>;
- fn flush_unsaved_buffer(
- self: &Client,
- stream_id: Identifier,
- topic_id: Identifier,
- partition_id: u32,
- fsync: bool,
- ) -> Result<()>;
fn get_stats(self: &Client) -> Result<Stats>;
fn get_me(self: &Client) -> Result<ClientInfoDetails>;
fn get_client(self: &Client, client_id: u32) ->
Result<ClientInfoDetails>;
diff --git a/foreign/cpp/tests/e2e/client.cpp b/foreign/cpp/tests/e2e/client.cpp
index b2b25f027..7d75b021b 100644
--- a/foreign/cpp/tests/e2e/client.cpp
+++ b/foreign/cpp/tests/e2e/client.cpp
@@ -2087,182 +2087,6 @@ TEST_F(LowLevelE2E_Client, GetStatsBeforeLoginThrows) {
ASSERT_THROW(client->get_stats(), std::exception);
}
-// The VSR server has no unsaved-buffer primitive (writes are journaled at
-// commit); FLUSH_UNSAVED_BUFFER denies typed with FeatureUnavailable even for
-// resolvable targets.
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferThrowsForExistingPartition) {
- RecordProperty("description",
- "Rejects flush_unsaved_buffer with the feature-unavailable
error for an existing partition.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- auto stream = client->get_stream(make_string_identifier(stream_name));
- TrackStream(stream.id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
- rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- messages.push_back(iggy::ffi::make_message(to_payload("flush-me"),
rust::Vec<iggy::ffi::HeaderEntry>()));
-
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id),
make_numeric_identifier(0),
- "partition_id",
partition_id_bytes(0), std::move(messages)));
-
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id),
make_numeric_identifier(0), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferThrowsForExistingEmptyPartition) {
- RecordProperty(
- "description",
- "Rejects flush_unsaved_buffer with the feature-unavailable error for a
partition with no unsaved messages.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- auto stream = client->get_stream(make_string_identifier(stream_name));
- TrackStream(stream.id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
-
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id),
make_numeric_identifier(0), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferBeforeLoginThrows) {
- RecordProperty("description",
- "Throws when flush_unsaved_buffer is called before connect,
after connect but before login, and "
- "after disconnect.");
- iggy::ffi::Client *client = GetLoggedOutClient();
-
- ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1),
make_numeric_identifier(1), 0, true),
- std::exception);
- ASSERT_NO_THROW(client->connect());
- ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1),
make_numeric_identifier(1), 0, true),
- std::exception);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(client->disconnect());
- ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1),
make_numeric_identifier(1), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferOnNonExistentStreamThrows) {
- RecordProperty("description", "Throws when flush_unsaved_buffer is called
for a stream that does not exist.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
- ASSERT_THROW(
- client->flush_unsaved_buffer(make_string_identifier(GetRandomName()),
make_numeric_identifier(0), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferOnNonExistentTopicThrows) {
- RecordProperty("description", "Throws when flush_unsaved_buffer is called
for a topic that does not exist.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
-
ASSERT_THROW(client->flush_unsaved_buffer(make_string_identifier(stream_name),
-
make_string_identifier(GetRandomName()), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferAfterStreamDeletedThrows) {
- RecordProperty("description", "Throws when flush_unsaved_buffer is called
after the stream has been deleted.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- auto stream = client->get_stream(make_string_identifier(stream_name));
- TrackStream(stream.id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
- const std::uint32_t saved_stream_id = stream.id;
-
ASSERT_NO_THROW(client->delete_stream(make_numeric_identifier(saved_stream_id)));
- ForgetTrackedStream(saved_stream_id);
-
- ASSERT_THROW(
- client->flush_unsaved_buffer(make_numeric_identifier(saved_stream_id),
make_numeric_identifier(0), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferAfterTopicDeletedThrows) {
- RecordProperty("description", "Throws when flush_unsaved_buffer is called
after the topic has been deleted.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- auto stream = client->get_stream(make_string_identifier(stream_name));
- TrackStream(stream.id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
- ASSERT_NO_THROW(client->delete_topic(make_numeric_identifier(stream.id),
make_string_identifier(topic_name)));
-
- ASSERT_THROW(
- client->flush_unsaved_buffer(make_numeric_identifier(stream.id),
make_string_identifier(topic_name), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferTwiceThrows) {
- RecordProperty("description",
- "Rejects flush_unsaved_buffer with the feature-unavailable
error consistently across repeat calls.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- auto stream = client->get_stream(make_string_identifier(stream_name));
- TrackStream(stream.id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
- rust::Vec<iggy::ffi::IggyMessageToSend> messages;
- messages.push_back(iggy::ffi::make_message(to_payload("flush-twice"),
rust::Vec<iggy::ffi::HeaderEntry>()));
-
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id),
make_numeric_identifier(0),
- "partition_id",
partition_id_bytes(0), std::move(messages)));
-
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id),
make_numeric_identifier(0), 0, true),
- std::exception);
-
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id),
make_numeric_identifier(0), 0, true),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_Client, FlushUnsavedBufferWithInvalidPartitionIdsThrows) {
- RecordProperty("description", "Throws when flush_unsaved_buffer is called
for non-existent partition ids.");
- const std::string stream_name = GetRandomName();
- const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
-
- ASSERT_NO_THROW(client->create_stream(stream_name));
- auto stream = client->get_stream(make_string_identifier(stream_name));
- TrackStream(stream.id);
- ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id),
topic_name,
- make_topic_create_options(1, "none",
"never_expire", 0, "server_default")));
-
- const std::uint32_t invalid_partition_ids[] = {1u, 9999u,
static_cast<std::uint32_t>(-1)};
- for (const std::uint32_t invalid_partition_id : invalid_partition_ids) {
- SCOPED_TRACE(invalid_partition_id);
-
ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id),
make_numeric_identifier(0),
- invalid_partition_id, true),
- std::exception);
- }
-}
-
TEST_F(LowLevelE2E_Client, DeleteSegmentsBeforeLoginThrows) {
RecordProperty("description",
"Rejects delete_segments before connect, after connect but
before login, and after disconnect.");
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/FlushMessagesTests.cs
b/foreign/csharp/Iggy_SDK.Tests.Integration/FlushMessagesTests.cs
deleted file mode 100644
index ea258700b..000000000
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/FlushMessagesTests.cs
+++ /dev/null
@@ -1,40 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-using Apache.Iggy.Enums;
-using Apache.Iggy.Exceptions;
-using Apache.Iggy.Tests.Integrations.Fixtures;
-using Shouldly;
-
-namespace Apache.Iggy.Tests.Integrations;
-
-public class FlushMessagesTests
-{
- [ClassDataSource<IggyServerFixture>(Shared = SharedType.PerAssembly)]
- public required IggyServerFixture Fixture { get; init; }
-
- [Test]
-
[MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
- public async Task
FlushUnsavedBuffer_Should_Throw_FeatureUnavailable(Protocol protocol)
- {
- var client = await Fixture.CreateAuthenticatedClient(protocol);
-
- await Should.ThrowAsync<FeatureUnavailableException>(() =>
- client.FlushUnsavedBufferAsync(Identifier.String("any-stream"),
- Identifier.String("any-topic"), 1, false));
- }
-}
diff --git a/foreign/csharp/Iggy_SDK/Contracts/Auth/GlobalPermissions.cs
b/foreign/csharp/Iggy_SDK/Contracts/Auth/GlobalPermissions.cs
index d56eeaab9..21dfeb353 100644
--- a/foreign/csharp/Iggy_SDK/Contracts/Auth/GlobalPermissions.cs
+++ b/foreign/csharp/Iggy_SDK/Contracts/Auth/GlobalPermissions.cs
@@ -193,9 +193,6 @@ public sealed class GlobalPermissions
/// <item>
/// <see
cref="Apache.Iggy.IggyClient.IIggyPublisher.SendMessagesAsync(Apache.Iggy.Identifier,
Apache.Iggy.Identifier, Apache.Iggy.Kinds.Partitioning,
System.Collections.Generic.IList{Apache.Iggy.Messages.Message},
System.Threading.CancellationToken)" />
/// </item>
- /// <item>
- /// <see
cref="Apache.Iggy.IggyClient.IIggyPublisher.FlushUnsavedBufferAsync" />
- /// </item>
/// </list>
/// </summary>
public required bool SendMessages { get; init; }
diff --git a/foreign/csharp/Iggy_SDK/Contracts/StreamPermissions.cs
b/foreign/csharp/Iggy_SDK/Contracts/StreamPermissions.cs
index 37011a7e9..00f15a0d6 100644
--- a/foreign/csharp/Iggy_SDK/Contracts/StreamPermissions.cs
+++ b/foreign/csharp/Iggy_SDK/Contracts/StreamPermissions.cs
@@ -134,9 +134,6 @@ public sealed class StreamPermissions
/// <item>
/// <see
cref="Apache.Iggy.IggyClient.IIggyPublisher.SendMessagesAsync(Apache.Iggy.Identifier,
Apache.Iggy.Identifier, Apache.Iggy.Kinds.Partitioning,
System.Collections.Generic.IList{Apache.Iggy.Messages.Message},
System.Threading.CancellationToken)" />
/// </item>
- /// <item>
- /// <see
cref="Apache.Iggy.IggyClient.IIggyPublisher.FlushUnsavedBufferAsync" />
- /// </item>
/// </list>
/// </summary>
public required bool SendMessages { get; init; }
diff --git a/foreign/csharp/Iggy_SDK/Contracts/TopicPermissions.cs
b/foreign/csharp/Iggy_SDK/Contracts/TopicPermissions.cs
index 1c68e4bb8..7ed97490f 100644
--- a/foreign/csharp/Iggy_SDK/Contracts/TopicPermissions.cs
+++ b/foreign/csharp/Iggy_SDK/Contracts/TopicPermissions.cs
@@ -102,9 +102,6 @@ public sealed class TopicPermissions
/// <item>
/// <see
cref="Apache.Iggy.IggyClient.IIggyPublisher.SendMessagesAsync(Apache.Iggy.Identifier,
Apache.Iggy.Identifier, Apache.Iggy.Kinds.Partitioning,
System.Collections.Generic.IList{Apache.Iggy.Messages.Message},
System.Threading.CancellationToken)" />
/// </item>
- /// <item>
- /// <see
cref="Apache.Iggy.IggyClient.IIggyPublisher.FlushUnsavedBufferAsync" />
- /// </item>
/// </list>
/// </summary>
public required bool SendMessages { get; init; }
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/IIggyPublisher.cs
b/foreign/csharp/Iggy_SDK/IggyClient/IIggyPublisher.cs
index 7263c220c..bf55d5442 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/IIggyPublisher.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/IIggyPublisher.cs
@@ -68,21 +68,4 @@ public interface IIggyPublisher
{
return SendMessagesAsync(streamId, topicId, partitioning, new[] {
message }, token);
}
-
- /// <summary>
- /// Forces a flush of the unsaved buffer to disk for a specific
partition.
- /// </summary>
- /// <remarks>
- /// This feature is not supported by the server. Durability is handled
by replication and the journal,
- /// so there is no client-flushable in-memory buffer.
- /// </remarks>
- /// <param name="streamId">The stream identifier (numeric ID or
name).</param>
- /// <param name="topicId">The topic identifier (numeric ID or
name).</param>
- /// <param name="partitionId">The partition identifier for which the
buffer should be flushed.</param>
- /// <param name="fsync">If true, the data is flushed and synchronized to
disk (fsync). If false, only flushed.</param>
- /// <param name="token">The cancellation token to cancel the
operation.</param>
- /// <returns>A task representing the asynchronous operation.</returns>
- /// <exception cref="Exceptions.FeatureUnavailableException">Always
thrown; the server does not support this command.</exception>
- Task FlushUnsavedBufferAsync(Identifier streamId, Identifier topicId, uint
partitionId, bool fsync,
- CancellationToken token = default);
}
diff --git
a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs
b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs
index baaa3adc3..5a3819420 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs
@@ -302,16 +302,6 @@ public class HttpMessageStream : IIggyClient
?? throw new InvalidResponseException("Send messages reply
carried no confirmation body.");
}
- /// <summary>
- /// This feature is not supported by the server.
- /// </summary>
- /// <exception cref="FeatureUnavailableException"></exception>
- public Task FlushUnsavedBufferAsync(Identifier streamId, Identifier
topicId, uint partitionId, bool fsync,
- CancellationToken token = default)
- {
- throw new FeatureUnavailableException();
- }
-
/// <inheritdoc />
public async Task<PolledMessages> PollMessagesAsync(Identifier streamId,
Identifier topicId, uint? partitionId,
Consumer consumer,
diff --git
a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index c68212a93..e14f2ec49 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -334,16 +334,6 @@ public sealed partial class TcpMessageStream : IIggyClient
return SendMessagesCoreAsync(streamId, topicId, partitioning, span,
token);
}
- /// <summary>
- /// This feature is not supported by the server.
- /// </summary>
- /// <exception cref="FeatureUnavailableException"></exception>
- public Task FlushUnsavedBufferAsync(Identifier streamId, Identifier
topicId, uint partitionId, bool fsync,
- CancellationToken token = default)
- {
- throw new FeatureUnavailableException();
- }
-
/// <inheritdoc />
public async Task<PolledMessages> PollMessagesAsync(Identifier streamId,
Identifier topicId, uint? partitionId,
Consumer consumer,
diff --git a/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs
b/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs
index 5ee64533e..76b5da491 100644
--- a/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs
+++ b/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs
@@ -45,7 +45,6 @@ internal static class CommandCodes
internal const int LOGIN_REGISTER_WITH_PAT_CODE = 45;
internal const int POLL_MESSAGES_CODE = 100;
internal const int SEND_MESSAGES_CODE = 101;
- internal const int FLUSH_UNSAVED_BUFFER_CODE = 102;
internal const int GET_POLL_ROUTING_CODE = 103;
internal const int POLL_MESSAGES_ON_PRIMARY_CODE = 104;
internal const int GET_CONSUMER_OFFSET_CODE = 120;
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs
b/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs
index b9d2d8ffe..814b5214f 100644
--- a/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs
@@ -71,8 +71,7 @@ internal static class VsrOperations
/// <summary>
/// Non-replicated codes this build knows to leave no server-side
state behind, so re-sending one after a
- /// lost connection is indistinguishable from sending it once.
Flushing an unsaved buffer is included: it
- /// is idempotent by construction, a second flush writes nothing new.
+ /// lost connection is indistinguishable from sending it once.
/// </summary>
private static readonly HashSet<int> NonReplicatedReadCodes =
[
@@ -86,7 +85,6 @@ internal static class VsrOperations
CommandCodes.GET_USER_CODE,
CommandCodes.GET_USERS_CODE,
CommandCodes.GET_PERSONAL_ACCESS_TOKENS_CODE,
- CommandCodes.FLUSH_UNSAVED_BUFFER_CODE,
CommandCodes.GET_CONSUMER_OFFSET_CODE,
CommandCodes.GET_STREAM_CODE,
CommandCodes.GET_STREAMS_CODE,
diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md
index b8c96ca7b..41feb6919 100644
--- a/foreign/csharp/README.md
+++ b/foreign/csharp/README.md
@@ -215,12 +215,13 @@ Replay-safe operations can also be retried after a lost
connection. Two cases su
and raw binary requests are TCP-only and throw `FeatureUnavailableException`
on HTTP.
- `StoreOffsetAsync` / `DeleteOffsetAsync` need an explicit partition id under
VSR: the broker does not
resolve a `null` partition for a consumer-offset request, so passing one
throws client-side.
-- `FlushUnsavedBufferAsync` always throws `FeatureUnavailableException`;
configure topic durability instead.
- Polling a missing topic throws `IggyInvalidStatusCodeException`. An existing
topic with no available
messages returns an empty poll.
### Behaviour changes for existing clients
+- `IIggyPublisher.FlushUnsavedBufferAsync` was removed. Set
`TopicOptions.Durability` to
+ `Durability.Persisted` when creating a topic to wait for message persistence.
- `MaxResponseFrameSize` bounds the reply frames the **VSR** reader accepts. A
reply larger than the 64 MiB
default is refused and the connection is dropped, so raise it if a single
response legitimately exceeds that
- a large `GetSnapshotAsync` is the usual case.
diff --git a/foreign/go/internal/vsr/protocol_parity_test.go
b/foreign/go/internal/vsr/protocol_parity_test.go
index be7f32ec7..0d8012183 100644
--- a/foreign/go/internal/vsr/protocol_parity_test.go
+++ b/foreign/go/internal/vsr/protocol_parity_test.go
@@ -146,10 +146,6 @@ var goHeaderOffsets = map[string]map[string]int{
},
}
-// unimplementedCommandCodes are protocol codes the Go SDK deliberately does
-// not declare. FlushUnsavedBuffer has no Go client method.
-var unimplementedCommandCodes = []uint32{102}
-
// rustFieldLayout is the size and alignment of every field type the consensus
// headers use, enough to recompute their repr(C) offsets.
var rustFieldLayout = map[string][2]int{
@@ -281,8 +277,7 @@ func TestProtocolParity_CommandCodes(t *testing.T) {
}
}
slices.Sort(missing)
- assert.Equal(t, unimplementedCommandCodes, missing,
- "the set of protocol codes the Go SDK does not declare has
changed")
+ assert.Empty(t, missing, "protocol command codes the Go SDK does not
declare")
}
func TestProtocolParity_OperationDiscriminants(t *testing.T) {
diff --git a/foreign/node/src/wire/command-set.ts
b/foreign/node/src/wire/command-set.ts
index ed8da501a..446ab2096 100644
--- a/foreign/node/src/wire/command-set.ts
+++ b/foreign/node/src/wire/command-set.ts
@@ -52,7 +52,6 @@ import { deleteOffset } from
'./offset/delete-offset.command.js';
import { sendMessages } from './message/send-messages.command.js';
import { pollMessages } from './message/poll-messages.command.js';
-import { flushUnsavedBuffers } from
'./message/flush-unsaved-buffers.command.js';
import { createStream } from './stream/create-stream.command.js';
import { updateStream } from './stream/update-stream.command.js';
@@ -182,8 +181,7 @@ type OffsetAPI = ReturnType<typeof offsetAPI>;
const messageAPI = (c: ClientProvider) => ({
poll: pollMessages(c),
- send: sendMessages(c),
- flushUnsavedBuffers: flushUnsavedBuffers(c)
+ send: sendMessages(c)
});
type MessageAPI = ReturnType<typeof messageAPI>;
diff --git a/foreign/node/src/wire/command.code.ts
b/foreign/node/src/wire/command.code.ts
index 9af84711f..4911489dd 100644
--- a/foreign/node/src/wire/command.code.ts
+++ b/foreign/node/src/wire/command.code.ts
@@ -44,7 +44,6 @@ export const COMMAND_CODE = {
LoginRegisterWithAccessToken: 45,
PollMessages: 100,
SendMessages: 101,
- FlushUnsavedBuffers: 102,
GetPollRouting: 103,
PollMessagesOnPrimary: 104,
GetOffset: 120,
diff --git a/foreign/node/src/wire/message/flush-unsaved-buffers.command.ts
b/foreign/node/src/wire/message/flush-unsaved-buffers.command.ts
deleted file mode 100644
index 3f8bff8d3..000000000
--- a/foreign/node/src/wire/message/flush-unsaved-buffers.command.ts
+++ /dev/null
@@ -1,53 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-import { type Id } from '../identifier.utils.js';
-import { serializeFlushUnsavedBuffers } from './message.utils.js';
-import { deserializeVoidResponse } from '../../client/client.utils.js';
-import { wrapCommand } from '../command.utils.js';
-import { COMMAND_CODE } from '../command.code.js';
-
-/**
- * Parameters for the flush unsaved buffers command.
- */
-export type FlushUnsavedBuffer = {
- /** Stream identifier */
- streamId: Id,
- /** Topic identifier */
- topicId: Id,
- /** Partition ID to flush */
- partitionId: number,
-};
-
-/**
- * Flush unsaved buffers command definition.
- * Forces buffered messages to be persisted to disk.
- */
-export const FLUSH_UNSAVED_BUFFERS = {
- code: COMMAND_CODE.FlushUnsavedBuffers,
-
- serialize: ({ streamId, topicId, partitionId }: FlushUnsavedBuffer) => {
- return serializeFlushUnsavedBuffers(streamId, topicId, partitionId);
- },
-
- deserialize: deserializeVoidResponse
-};
-
-/**
- * Executable flush unsaved buffers command function.
- */
-export const flushUnsavedBuffers = wrapCommand<FlushUnsavedBuffer,
boolean>(FLUSH_UNSAVED_BUFFERS);
diff --git a/foreign/node/src/wire/message/message.utils.ts
b/foreign/node/src/wire/message/message.utils.ts
index 576dd01d4..a002a9d34 100644
--- a/foreign/node/src/wire/message/message.utils.ts
+++ b/foreign/node/src/wire/message/message.utils.ts
@@ -16,7 +16,7 @@
// under the License.
import { randomFillSync } from 'node:crypto';
-import { uint32ToBuf, u128ToBuf, uint8ToBuf } from '../number.utils.js';
+import { uint32ToBuf, u128ToBuf } from '../number.utils.js';
import { serializeHeaders, type Headers } from './header.utils.js';
import { serializeIdentifier, type Id } from '../identifier.utils.js';
import { serializePartitioning, type Partitioning } from
'./partitioning.utils.js';
@@ -291,31 +291,3 @@ export const serializeSendMessages = (
bBatch
]);
};
-
-/**
- * Serializes a flush unsaved buffers command payload.
- *
- * @param streamId - Stream identifier
- * @param topicId - Topic identifier
- * @param partitionId - Partition ID to flush
- * @param fsync - Whether to force sync to disk
- * @returns Serialized command payload
- */
-export const serializeFlushUnsavedBuffers = (
- streamId: Id,
- topicId: Id,
- partitionId: number,
- fsync = false
-) => {
- const streamIdentifier = serializeIdentifier(streamId);
- const topicIdentifier = serializeIdentifier(topicId);
- const bPartitionId = uint32ToBuf(partitionId);
- const bFSync = uint8ToBuf(fsync ? 1 : 0);
-
- return Buffer.concat([
- streamIdentifier,
- topicIdentifier,
- bPartitionId,
- bFSync
- ]);
-};
