diff --git a/bdd/cpp/features/step_definitions/messaging_steps.cpp b/bdd/cpp/features/step_definitions/messaging_steps.cpp index 90ee3ba7e4..9f5524dd19 100644 --- a/bdd/cpp/features/step_definitions/messaging_steps.cpp +++ b/bdd/cpp/features/step_definitions/messaging_steps.cpp @@ -150,7 +150,7 @@ WHEN("^I poll messages from stream ([0-9]+), topic ([0-9]+), partition ([0-9]+) const auto polled = context->client->poll_messages( bdd::make_numeric_identifier(static_cast(stream_id)), bdd::make_numeric_identifier(static_cast(topic_id)), static_cast(partition_id), - "consumer", bdd::make_numeric_identifier(1), std::string(polling_strategy.PollingStrategyKind()), + iggy::Consumer::Single(1), std::string(polling_strategy.PollingStrategyKind()), polling_strategy.PollingStrategyValue(), 100, false); context->polled.count = polled.count; diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp index 593815299e..25364878d9 100644 --- a/foreign/cpp/include/iggy.hpp +++ b/foreign/cpp/include/iggy.hpp @@ -485,6 +485,117 @@ class IggyException : public std::runtime_error { explicit IggyException(const std::string &message) : std::runtime_error(message) {} }; +/** + * @brief Partition value that names no partition. + * + * Polling a consumer group with it reads one of the partitions assigned to the + * polling member, taking the next one on every call. Polling a regular consumer + * with it reads partition 0, and so does `get_consumer_offset(...)`. + * `store_consumer_offset(...)` and `delete_consumer_offset(...)` reject it and + * need an explicit partition. + */ +inline constexpr std::uint32_t kAnyPartitionId{std::numeric_limits::max()}; + +namespace detail { + +inline iggy::ffi::Identifier to_string_identifier(const std::string &id) { + iggy::ffi::Identifier identifier{}; + try { + identifier.set_string(id); + } catch (const std::exception &error) { + throw IggyException(error.what()); + } + + return identifier; +} + +inline iggy::ffi::Identifier to_numeric_identifier(const std::uint32_t id) { + iggy::ffi::Identifier identifier{}; + try { + identifier.set_numeric(id); + } catch (const std::exception &error) { + throw IggyException(error.what()); + } + + return identifier; +} + +inline iggy::ffi::Consumer to_consumer(const iggy::ffi::ConsumerKind kind, iggy::ffi::Identifier id) { + iggy::ffi::Consumer consumer{}; + consumer.kind = kind; + consumer.id = std::move(id); + + return consumer; +} + +} // namespace detail + +/** + * @brief Creates the consumer that polling and consumer-offset calls take. + * + * The consumer carries the kind and the identifier the server keys the stored + * offset on. Two callers naming the same identifier share one offset, so under + * `PollingStrategy::Next()` with auto-commit each of them sees only the messages + * the other has not read yet. Give every independent consumer its own name. + * + * @code{.cpp} + * const auto strategy{iggy::PollingStrategy::Next()}; + * const auto polled{client->poll_messages(stream, topic, iggy::kAnyPartitionId, + * iggy::Consumer::Group("my-group"), + * std::string(strategy.PollingStrategyKind()), + * strategy.PollingStrategyValue(), 10, true)}; + * @endcode + */ +class Consumer final { + public: + /** + * @brief Creates a named consumer owning its offset on the polled partition. + * @param id Consumer name. + * @return Consumer accepted by the client. + * @throws IggyException if @p id is empty or longer than 255 bytes. + */ + static iggy::ffi::Consumer Single(const std::string &id) { + return detail::to_consumer(iggy::ffi::ConsumerKind::Consumer, detail::to_string_identifier(id)); + } + + /** + * @brief Creates a numbered consumer owning its offset on the polled partition. + * @param id Consumer number. + * @return Consumer accepted by the client. + */ + static iggy::ffi::Consumer Single(const std::uint32_t id) { + return detail::to_consumer(iggy::ffi::ConsumerKind::Consumer, detail::to_numeric_identifier(id)); + } + + /** + * @brief Creates a member of the named consumer group, sharing the group's offset. + * + * Join the group with `join_consumer_group(...)` before polling, and pass + * `kAnyPartitionId` to read the partitions assigned to this member, one per + * call. + * + * @param id Consumer group name. + * @return Consumer accepted by the client. + * @throws IggyException if @p id is empty or longer than 255 bytes. + */ + static iggy::ffi::Consumer Group(const std::string &id) { + return detail::to_consumer(iggy::ffi::ConsumerKind::ConsumerGroup, detail::to_string_identifier(id)); + } + + /** + * @brief Creates a member of the numbered consumer group, sharing the group's offset. + * @param id Consumer group number. + * @return Consumer accepted by the client. + * @see Group(const std::string &) + */ + static iggy::ffi::Consumer Group(const std::uint32_t id) { + return detail::to_consumer(iggy::ffi::ConsumerKind::ConsumerGroup, detail::to_numeric_identifier(id)); + } + + private: + Consumer() = delete; +}; + /** * @brief Owning client connection to an Apache Iggy server. * diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs index 002ead54da..1792c9f3d3 100644 --- a/foreign/cpp/src/client.rs +++ b/foreign/cpp/src/client.rs @@ -42,14 +42,6 @@ use std::sync::Arc; /// reserve `u32::MAX` as the sentinel for `partition_id`. const ANY_PARTITION_ID: u32 = u32::MAX; -fn resolve_consumer(consumer_kind: &str, consumer_id: RustIdentifier) -> Result { - match consumer_kind { - "consumer" => Ok(Consumer::new(consumer_id)), - "consumer_group" => Ok(Consumer::group(consumer_id)), - _ => Err(format!("invalid consumer kind: {consumer_kind}")), - } -} - fn opt_partition(partition_id: u32) -> Option { if partition_id == ANY_PARTITION_ID { None @@ -391,8 +383,7 @@ impl Client { stream_id: ffi::Identifier, topic_id: ffi::Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: ffi::Identifier, + consumer: ffi::Consumer, polling_strategy_kind: String, polling_strategy_value: u64, count: u32, @@ -402,9 +393,7 @@ impl Client { .map_err(|error| format!("Could not poll messages: {error}"))?; let rust_topic_id = RustIdentifier::try_from(topic_id) .map_err(|error| format!("Could not poll messages: {error}"))?; - let rust_consumer_id = RustIdentifier::try_from(consumer_id) - .map_err(|error| format!("Could not poll messages: {error}"))?; - let consumer = resolve_consumer(&consumer_kind, rust_consumer_id) + let rust_consumer = Consumer::try_from(consumer) .map_err(|error| format!("Could not poll messages: {error}"))?; let strategy = match polling_strategy_kind.as_str() { @@ -427,7 +416,7 @@ impl Client { &rust_stream_id, &rust_topic_id, opt_partition(partition_id), - &consumer, + &rust_consumer, &strategy, count, auto_commit, @@ -911,23 +900,20 @@ impl Client { stream_id: ffi::Identifier, topic_id: ffi::Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: ffi::Identifier, + consumer: ffi::Consumer, offset: u64, ) -> Result<(), String> { let rust_stream_id = RustIdentifier::try_from(stream_id) .map_err(|error| format!("Could not store consumer offset: {error}"))?; let rust_topic_id = RustIdentifier::try_from(topic_id) .map_err(|error| format!("Could not store consumer offset: {error}"))?; - let rust_consumer_id = RustIdentifier::try_from(consumer_id) - .map_err(|error| format!("Could not store consumer offset: {error}"))?; - let consumer = resolve_consumer(&consumer_kind, rust_consumer_id) + let rust_consumer = Consumer::try_from(consumer) .map_err(|error| format!("Could not store consumer offset: {error}"))?; RUNTIME.block_on(async { self.inner .store_consumer_offset( - &consumer, + &rust_consumer, &rust_stream_id, &rust_topic_id, opt_partition(partition_id), @@ -948,23 +934,20 @@ impl Client { stream_id: ffi::Identifier, topic_id: ffi::Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: ffi::Identifier, + consumer: ffi::Consumer, ) -> Result { let rust_stream_id = RustIdentifier::try_from(stream_id) .map_err(|error| format!("Could not get consumer offset: {error}"))?; let rust_topic_id = RustIdentifier::try_from(topic_id) .map_err(|error| format!("Could not get consumer offset: {error}"))?; - let rust_consumer_id = RustIdentifier::try_from(consumer_id) - .map_err(|error| format!("Could not get consumer offset: {error}"))?; - let consumer = resolve_consumer(&consumer_kind, rust_consumer_id) + let rust_consumer = Consumer::try_from(consumer) .map_err(|error| format!("Could not get consumer offset: {error}"))?; RUNTIME.block_on(async { let offset = self .inner .get_consumer_offset( - &consumer, + &rust_consumer, &rust_stream_id, &rust_topic_id, opt_partition(partition_id), @@ -988,22 +971,19 @@ impl Client { stream_id: ffi::Identifier, topic_id: ffi::Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: ffi::Identifier, + consumer: ffi::Consumer, ) -> Result<(), String> { let rust_stream_id = RustIdentifier::try_from(stream_id) .map_err(|error| format!("Could not delete consumer offset: {error}"))?; let rust_topic_id = RustIdentifier::try_from(topic_id) .map_err(|error| format!("Could not delete consumer offset: {error}"))?; - let rust_consumer_id = RustIdentifier::try_from(consumer_id) - .map_err(|error| format!("Could not delete consumer offset: {error}"))?; - let consumer = resolve_consumer(&consumer_kind, rust_consumer_id) + let rust_consumer = Consumer::try_from(consumer) .map_err(|error| format!("Could not delete consumer offset: {error}"))?; RUNTIME.block_on(async { self.inner .delete_consumer_offset( - &consumer, + &rust_consumer, &rust_stream_id, &rust_topic_id, opt_partition(partition_id), diff --git a/foreign/cpp/src/consumer.rs b/foreign/cpp/src/consumer.rs index 147374d4ed..1d0be609ad 100644 --- a/foreign/cpp/src/consumer.rs +++ b/foreign/cpp/src/consumer.rs @@ -18,6 +18,6 @@ use iggy::prelude::IggyConsumer as RustIggyConsumer; #[allow(dead_code)] -pub struct Consumer { +pub struct IggyConsumer { pub inner: RustIggyConsumer, } diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs index 496b0728fc..20855b0396 100644 --- a/foreign/cpp/src/lib.rs +++ b/foreign/cpp/src/lib.rs @@ -23,7 +23,7 @@ mod producer; mod type_conversion; use client::{Client, delete_connection as delete_client, from_connection_string, new_connection}; -use consumer::Consumer; +use consumer::IggyConsumer; use messages::make_message; use producer::Producer; use std::sync::LazyLock; @@ -244,6 +244,22 @@ mod ffi { stored_offset: u64, } + /// Kind of the consumer a poll or a consumer-offset call names. + #[repr(u8)] + enum ConsumerKind { + Consumer = 1, + ConsumerGroup = 2, + } + + /// The consumer a poll or a consumer-offset call names. It carries both the + /// kind and the identifier the server keys the stored offset on. Two callers + /// sharing one identifier share one offset, so under the `next` strategy with + /// auto-commit each of them reads only what the other has not read yet. + struct Consumer { + kind: ConsumerKind, + id: Identifier, + } + struct ClientInfo { client_id: u32, has_user_id: bool, @@ -430,7 +446,7 @@ mod ffi { extern "Rust" { type Client; - type Consumer; + type IggyConsumer; type Producer; // Client functions @@ -529,8 +545,7 @@ mod ffi { stream_id: Identifier, topic_id: Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: Identifier, + consumer: Consumer, offset: u64, ) -> Result<()>; fn get_consumer_offset( @@ -538,26 +553,27 @@ mod ffi { stream_id: Identifier, topic_id: Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: Identifier, + consumer: Consumer, ) -> Result; fn delete_consumer_offset( self: &Client, stream_id: Identifier, topic_id: Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: Identifier, + consumer: Consumer, ) -> Result<()>; + /// Polls messages as `consumer`. A `partition_id` of `u32::MAX` names no + /// partition: a consumer group then reads one of the partitions assigned + /// to the polling member, taking the next one on every call, and a + /// regular consumer reads partition 0. #[allow(clippy::too_many_arguments)] fn poll_messages( self: &Client, stream_id: Identifier, topic_id: Identifier, partition_id: u32, - consumer_kind: String, - consumer_id: Identifier, + consumer: Consumer, polling_strategy_kind: String, polling_strategy_value: u64, count: u32, @@ -657,18 +673,18 @@ mod ffi { fn set_string(self: &mut Identifier, id: String) -> Result<()>; fn set_numeric(self: &mut Identifier, id: u32) -> Result<()>; - // Consumer methods - // fn name(self: &Consumer) -> Result; - // fn topic(self: &Consumer) -> Result; - // fn stream(self: &Consumer) -> Result; - // fn partition_id(self: &Consumer) -> u32; - // fn store_offset(self: &Consumer, offset: u64, partition_id: u32) -> Result<()>; - // fn delete_offset(self: &Consumer, partition_id: u32) -> Result<()>; - // fn get_last_consumed_offset(self: &Consumer, partition_id: u32) -> Result; - // fn get_last_stored_offset(self: &Consumer, partition_id: u32) -> Result; - // fn init(self: &mut Consumer) -> Result<()>; - // fn shutdown(self: &mut Consumer) -> Result<()>; - // unsafe fn delete_consumer(consumer: *mut Consumer) -> Result<()>; + // IggyConsumer methods + // fn name(self: &IggyConsumer) -> Result; + // fn topic(self: &IggyConsumer) -> Result; + // fn stream(self: &IggyConsumer) -> Result; + // fn partition_id(self: &IggyConsumer) -> u32; + // fn store_offset(self: &IggyConsumer, offset: u64, partition_id: u32) -> Result<()>; + // fn delete_offset(self: &IggyConsumer, partition_id: u32) -> Result<()>; + // fn get_last_consumed_offset(self: &IggyConsumer, partition_id: u32) -> Result; + // fn get_last_stored_offset(self: &IggyConsumer, partition_id: u32) -> Result; + // fn init(self: &mut IggyConsumer) -> Result<()>; + // fn shutdown(self: &mut IggyConsumer) -> Result<()>; + // unsafe fn delete_consumer(consumer: *mut IggyConsumer) -> Result<()>; // Producer methods // fn stream(self: &Producer) -> Result; diff --git a/foreign/cpp/src/type_conversion.rs b/foreign/cpp/src/type_conversion.rs index 83d4c52b6d..ebbfc117b4 100644 --- a/foreign/cpp/src/type_conversion.rs +++ b/foreign/cpp/src/type_conversion.rs @@ -18,7 +18,8 @@ use crate::ffi; use bytes::Bytes; use iggy::prelude::{ - ConsumerGroupDetails as RustConsumerGroupDetails, IdKind, Identifier as RustIdentifier, + Consumer as RustConsumer, ConsumerGroupDetails as RustConsumerGroupDetails, + ConsumerKind as RustConsumerKind, IdKind, Identifier as RustIdentifier, IggyMessage as RustIggyMessage, OptionSpec as RustOptionSpec, Partition as RustPartition, PolledMessages as RustPolledMessages, SendMessagesConfirmationResponse as RustSendMessagesConfirmationResponse, @@ -85,6 +86,29 @@ impl TryFrom for RustIdentifier { } } +impl TryFrom for RustConsumerKind { + type Error = String; + + fn try_from(kind: ffi::ConsumerKind) -> Result { + match kind { + ffi::ConsumerKind::Consumer => Ok(RustConsumerKind::Consumer), + ffi::ConsumerKind::ConsumerGroup => Ok(RustConsumerKind::ConsumerGroup), + _ => Err(format!("unsupported consumer kind '{}'", kind.repr)), + } + } +} + +impl TryFrom for RustConsumer { + type Error = String; + + fn try_from(consumer: ffi::Consumer) -> Result { + Ok(RustConsumer { + kind: RustConsumerKind::try_from(consumer.kind)?, + id: RustIdentifier::try_from(consumer.id)?, + }) + } +} + impl From for ffi::ClientInfo { fn from(client: RustClientInfo) -> Self { let has_user_id = client.user_id.is_some(); diff --git a/foreign/cpp/tests/e2e/client.cpp b/foreign/cpp/tests/e2e/client.cpp index 5fde598711..42dcc3e2f3 100644 --- a/foreign/cpp/tests/e2e/client.cpp +++ b/foreign/cpp/tests/e2e/client.cpp @@ -17,8 +17,6 @@ * under the License. */ -// TODO(slbotbm): Add tests for store_consumer_offset, get_consumer_offset, and delete_consumer_offset functions -// attached to client after implementing consumer group functions // TODO(slbotbm): Add tests for update_permissions after creating create_user, get_user, etc. functions #include #include @@ -31,6 +29,7 @@ #include +#include "iggy.hpp" #include "lib.rs.h" #include "tests/e2e/test_helpers.hpp" @@ -2384,8 +2383,8 @@ TEST_F(LowLevelE2E_Client, DeleteSegmentsWithZeroCountIsNoOp) { iggy::ffi::PolledMessages polled_before_delete{}; ASSERT_NO_THROW({ polled_before_delete = - client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer", - make_numeric_identifier(1005), "offset", 0, 1000, false); + client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, + iggy::Consumer::Single(1005), "offset", 0, 1000, false); }); ASSERT_NO_THROW( @@ -2406,8 +2405,8 @@ TEST_F(LowLevelE2E_Client, DeleteSegmentsWithZeroCountIsNoOp) { iggy::ffi::PolledMessages polled_after_delete{}; ASSERT_NO_THROW({ polled_after_delete = - client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer", - make_numeric_identifier(1006), "offset", 0, 1000, false); + client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, + iggy::Consumer::Single(1006), "offset", 0, 1000, false); }); EXPECT_EQ(partition_after_delete.segments_count, partition_before_delete.segments_count); @@ -2466,8 +2465,8 @@ TEST_F(LowLevelE2E_Client, DeleteSegmentsWhenOnlyActiveSegmentRemainsIsNoOp) { iggy::ffi::PolledMessages polled_before_delete{}; ASSERT_NO_THROW({ polled_before_delete = - client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer", - make_numeric_identifier(1007), "offset", 0, 1000, false); + client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, + iggy::Consumer::Single(1007), "offset", 0, 1000, false); }); ASSERT_NO_THROW( @@ -2488,8 +2487,8 @@ TEST_F(LowLevelE2E_Client, DeleteSegmentsWhenOnlyActiveSegmentRemainsIsNoOp) { iggy::ffi::PolledMessages polled_after_delete{}; ASSERT_NO_THROW({ polled_after_delete = - client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer", - make_numeric_identifier(1008), "offset", 0, 1000, false); + client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, + iggy::Consumer::Single(1008), "offset", 0, 1000, false); }); EXPECT_EQ(partition_after_delete.segments_count, partition_before_delete.segments_count); @@ -3287,3 +3286,92 @@ TEST_F(LowLevelE2E_Client, SendBinaryRequestUnknownCommandCodeThrows) { rust::Vec empty_payload; ASSERT_THROW(client->send_binary_request(unknown_command_code, empty_payload), std::exception); } + +TEST_F(LowLevelE2E_Client, ConsumerOffsetStoreGetDeleteRoundTrip) { + RecordProperty("description", "Stores a consumer offset, reads it back, and deletes it."); + const std::string stream_name = GetRandomName(); + iggy::ffi::Client *client = GetLoggedInClient(); + + client->create_stream(stream_name); + auto stream = client->get_stream(make_string_identifier(stream_name)); + TrackStream(stream.id); + const std::string topic_name = GetRandomName(); + client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", + {}); + + rust::Vec messages; + for (std::uint32_t i = 0; i < 5; i++) { + auto msg = + iggy::ffi::make_message(to_payload("offset-" + std::to_string(i)), rust::Vec()); + messages.push_back(std::move(msg)); + } + client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", + partition_id_bytes(0), std::move(messages)); + + ASSERT_NO_THROW(client->store_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("offset-consumer"), 2)); + + auto stored = client->get_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("offset-consumer")); + EXPECT_EQ(stored.partition_id, 0u); + EXPECT_EQ(stored.stored_offset, 2u); + + ASSERT_NO_THROW(client->delete_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("offset-consumer"))); + + ASSERT_THROW(client->get_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("offset-consumer")), + std::exception); +} + +TEST_F(LowLevelE2E_Client, ConsumerOffsetKeepsDistinctConsumersApart) { + RecordProperty("description", "Each consumer owns its stored offset on the same partition."); + const std::string stream_name = GetRandomName(); + iggy::ffi::Client *client = GetLoggedInClient(); + + client->create_stream(stream_name); + auto stream = client->get_stream(make_string_identifier(stream_name)); + TrackStream(stream.id); + const std::string topic_name = GetRandomName(); + client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", + {}); + + rust::Vec messages; + for (std::uint32_t i = 0; i < 5; i++) { + auto msg = + iggy::ffi::make_message(to_payload("offset-" + std::to_string(i)), rust::Vec()); + messages.push_back(std::move(msg)); + } + client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", + partition_id_bytes(0), std::move(messages)); + + client->store_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("consumer-a"), 1); + client->store_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("consumer-b"), 3); + + auto offset_a = client->get_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("consumer-a")); + auto offset_b = client->get_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("consumer-b")); + + EXPECT_EQ(offset_a.stored_offset, 1u); + EXPECT_EQ(offset_b.stored_offset, 3u); +} + +TEST_F(LowLevelE2E_Client, StoreConsumerOffsetRejectsAnyPartitionId) { + RecordProperty("description", "Storing an offset needs an explicit partition, unlike polling."); + const std::string stream_name = GetRandomName(); + iggy::ffi::Client *client = GetLoggedInClient(); + + client->create_stream(stream_name); + auto stream = client->get_stream(make_string_identifier(stream_name)); + TrackStream(stream.id); + const std::string topic_name = GetRandomName(); + client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", + {}); + + ASSERT_THROW(client->store_consumer_offset(make_numeric_identifier(stream.id), make_numeric_identifier(0), + iggy::kAnyPartitionId, iggy::Consumer::Single("offset-consumer"), 1), + std::exception); +} diff --git a/foreign/cpp/tests/e2e/message.cpp b/foreign/cpp/tests/e2e/message.cpp index 09b1e2ae1f..ca80799a66 100644 --- a/foreign/cpp/tests/e2e/message.cpp +++ b/foreign/cpp/tests/e2e/message.cpp @@ -20,9 +20,12 @@ #include #include #include +#include +#include #include +#include "iggy.hpp" #include "lib.rs.h" #include "tests/e2e/test_helpers.hpp" @@ -56,8 +59,8 @@ TEST_F(LowLevelE2E_Message, SendAndPollMessagesRoundTrip) { << "must carry exactly one confirmation"; EXPECT_EQ(sent.confirmations.front().partition_id, 0u); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.partition_id, 0u) << "Polled partition_id mismatches the partition we sent to"; ASSERT_EQ(polled.count, 10u); @@ -91,8 +94,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesVerifyMessageIds) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.messages.size(), 1u); ASSERT_EQ(polled.messages[0].id_lo, 42u); @@ -111,8 +114,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesFromEmptyPartition) { client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", {}); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.count, 0u); ASSERT_EQ(polled.messages.size(), 0u); @@ -242,12 +245,12 @@ TEST_F(LowLevelE2E_Message, SendMessagesToSpecificPartitionVerified) { partition_id_bytes(0), std::move(messages)); auto polled_part0 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, - "consumer", make_numeric_identifier(1), "offset", 0, 100, false); + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled_part0.partition_id, 0u); ASSERT_EQ(polled_part0.count, 5u); auto polled_part1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 1, - "consumer", make_numeric_identifier(1), "offset", 0, 100, false); + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled_part1.partition_id, 1u); ASSERT_EQ(polled_part1.count, 0u); } @@ -344,8 +347,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesPreservesOrder) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.count, 50u); for (std::uint32_t i = 0; i < 50; i++) { @@ -380,8 +383,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesWithDuplicateIds) { ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages))); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.count, 3u); for (std::size_t i = 0; i < polled.messages.size(); i++) { @@ -431,8 +434,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesWithVariousPayloads) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.count, 4u); @@ -525,8 +528,8 @@ TEST_F(LowLevelE2E_Message, SendAndPollMessageWithTypedHeadersRoundTrip) { ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages))); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); constexpr std::size_t expected_message_count = sizeof(expected_messages) / sizeof(expected_messages[0]); ASSERT_EQ(polled.count, expected_message_count); @@ -793,8 +796,8 @@ TEST_F(LowLevelE2E_Message, SendMessageAtUserHeadersSizeBoundary) { ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(exact_messages))); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 10, false); ASSERT_EQ(polled.count, 1u); ASSERT_EQ(polled.messages.size(), 1u); @@ -809,13 +812,13 @@ TEST_F(LowLevelE2E_Message, PollMessagesBeforeLoginThrows) { iggy::ffi::Client *client = GetLoggedOutClient(); ASSERT_NO_THROW(client->connect()); - ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 10, false), std::exception); ASSERT_NO_THROW(client->login_user("iggy", "iggy")); ASSERT_NO_THROW(client->disconnect()); - ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 10, false), std::exception); } @@ -827,8 +830,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidStreamIdThrows) { invalid_id.kind = "invalid"; invalid_id.length = 0; - ASSERT_THROW(client->poll_messages(invalid_id, make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false), + ASSERT_THROW(client->poll_messages(invalid_id, make_numeric_identifier(0), 0, iggy::Consumer::Single(1), "offset", + 0, 10, false), std::exception); } @@ -837,7 +840,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesFromNonExistentStreamThrows) { iggy::ffi::Client *client = GetLoggedInClient(); ASSERT_THROW(client->poll_messages(make_string_identifier("nonexistent-stream-poll"), make_numeric_identifier(0), 0, - "consumer", make_numeric_identifier(1), "offset", 0, 10, false), + iggy::Consumer::Single(1), "offset", 0, 10, false), std::exception); } @@ -853,8 +856,12 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidConsumerKindThrows) { client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", {}); - ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "invalid", - make_numeric_identifier(1), "offset", 0, 10, false), + iggy::ffi::Consumer consumer{}; + consumer.kind = static_cast(0); + consumer.id = make_numeric_identifier(1); + + ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + std::move(consumer), "offset", 0, 10, false), std::exception); } @@ -870,8 +877,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidStrategyKindThrows) { client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", {}); - ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "invalid", 0, 10, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "invalid", 0, 10, false), std::exception); } @@ -896,8 +903,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesCountLessThanAvailable) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 5, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 5, false); ASSERT_EQ(polled.count, 5u); ASSERT_EQ(polled.messages.size(), 5u); @@ -924,8 +931,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithLargeOffset) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 999999, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 999999, 100, false); ASSERT_EQ(polled.count, 0u); ASSERT_EQ(polled.messages.size(), 0u); @@ -952,8 +959,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesFirstStrategy) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "first", 0, 3, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "first", 0, 3, false); ASSERT_EQ(polled.count, 3u); ASSERT_EQ(polled.messages.size(), 3u); @@ -987,8 +994,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesLastStrategy) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "last", 0, 3, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "last", 0, 3, false); ASSERT_EQ(polled.count, 3u); ASSERT_EQ(polled.messages.size(), 3u); @@ -1023,12 +1030,12 @@ TEST_F(LowLevelE2E_Message, PollMessagesNextStrategyNoAutoCommit) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "next", 0, 100, false); + auto polled1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "next", 0, 100, false); ASSERT_EQ(polled1.count, 5u); - auto polled2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "next", 0, 100, false); + auto polled2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "next", 0, 100, false); ASSERT_EQ(polled2.count, 5u); for (std::uint32_t i = 0; i < 5; i++) { EXPECT_EQ(polled1.messages[i].offset, static_cast(i)); @@ -1065,14 +1072,14 @@ TEST_F(LowLevelE2E_Message, PollMessagesNextStrategyAutoCommit) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "next", 0, 5, true); + auto polled1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "next", 0, 5, true); ASSERT_EQ(polled1.count, 5u); EXPECT_EQ(polled1.messages[0].offset, 0u); EXPECT_EQ(polled1.messages[4].offset, 4u); - auto polled2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "next", 0, 5, true); + auto polled2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "next", 0, 5, true); ASSERT_EQ(polled2.count, 5u); EXPECT_EQ(polled2.messages[0].offset, 5u); EXPECT_EQ(polled2.messages[4].offset, 9u); @@ -1087,8 +1094,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesNextStrategyAutoCommit) { EXPECT_EQ(actual2, expected2) << "polled2 payload mismatch at index " << i; } - auto polled3 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "next", 0, 5, true); + auto polled3 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "next", 0, 5, true); ASSERT_EQ(polled3.count, 0u); } @@ -1114,15 +1121,15 @@ TEST_F(LowLevelE2E_Message, PollMessagesConsumerIdIndependence) { partition_id_bytes(0), std::move(messages)); auto polled_c1 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, - "consumer", make_numeric_identifier(1), "next", 0, 3, true); + iggy::Consumer::Single(1), "next", 0, 3, true); ASSERT_EQ(polled_c1.count, 3u); auto polled_c2 = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, - "consumer", make_numeric_identifier(2), "next", 0, 5, true); + iggy::Consumer::Single(2), "next", 0, 5, true); ASSERT_EQ(polled_c2.count, 5u); auto polled_c1_again = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, - "consumer", make_numeric_identifier(1), "next", 0, 5, true); + iggy::Consumer::Single(1), "next", 0, 5, true); ASSERT_EQ(polled_c1_again.count, 2u); } @@ -1156,8 +1163,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesMultipleSendsThenPollOrder) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(batch2)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.count, 10u); for (std::uint32_t i = 0; i < 10; i++) { @@ -1199,8 +1206,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesMultipleCustomIds) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages)); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(polled.count, 5u); for (std::uint32_t i = 0; i < 5; i++) { @@ -1233,7 +1240,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesAfterStreamDeletedThrows) { ForgetTrackedStream(saved_stream_id); ASSERT_THROW(client->poll_messages(make_numeric_identifier(saved_stream_id), make_numeric_identifier(0), 0, - "consumer", make_numeric_identifier(1), "offset", 0, 10, false), + iggy::Consumer::Single(1), "offset", 0, 10, false), std::exception); } @@ -1249,8 +1256,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidPartitionIdThrows) { client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", {}); - ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 9999, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 9999, + iggy::Consumer::Single(1), "offset", 0, 10, false), std::exception); } @@ -1266,8 +1273,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithCountZeroThrows) { client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", {}); - ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 0, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 0, false), std::exception); } @@ -1293,7 +1300,7 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithoutSpecifyingPartition) { partition_id_bytes(0), std::move(messages)); auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), UINT32_MAX, - "consumer", make_numeric_identifier(1), "offset", 0, 100, false); + iggy::Consumer::Single(1), "offset", 0, 100, false); // The Rust side maps UINT32_MAX to None, so the server picks a partition. With a single // partition topic that should always be partition 0. @@ -1340,8 +1347,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesTimestampStrategy) { client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(batch2)); - auto all = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 100, false); + auto all = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 100, false); ASSERT_EQ(all.count, 10u); // IggyTimestamp::now() is microsecond-resolution and we slept 100ms between batches; a gap @@ -1354,8 +1361,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesTimestampStrategy) { << "Timestamp gap collapsed (" << (batch2_timestamp - batch1_timestamp) << "us) — test no longer exercises timestamp filtering"; - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(2), "timestamp", batch2_timestamp, 100, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(2), "timestamp", batch2_timestamp, 100, false); ASSERT_GE(polled.count, 5u); // The server contract is `timestamp >= polling_strategy_value`. If a batch1 message lands on @@ -1394,9 +1401,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesMonotonicOffsets) { std::uint64_t expected_offset = 0; for (int chunk = 0; chunk < 4; chunk++) { - auto polled = - client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", expected_offset, 5, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", expected_offset, 5, false); ASSERT_EQ(polled.count, 5u) << "Chunk " << chunk; ASSERT_EQ(polled.messages.size(), 5u) << "Chunk " << chunk; @@ -1432,8 +1438,8 @@ TEST_F(LowLevelE2E_Message, SendMessagesLargeBatch) { ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", partition_id_bytes(0), std::move(messages))); - auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 1000, false); + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(1), "offset", 0, 1000, false); ASSERT_EQ(polled.count, 1000u); ASSERT_EQ(polled.messages.size(), 1000u); @@ -1466,8 +1472,8 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidTopicIdThrows) { invalid_id.kind = "invalid"; invalid_id.length = 0; - ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), invalid_id, 0, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(1), invalid_id, 0, iggy::Consumer::Single(1), "offset", + 0, 10, false), std::exception); } @@ -1483,12 +1489,13 @@ TEST_F(LowLevelE2E_Message, PollMessagesWithInvalidConsumerIdThrows) { client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", {}); - iggy::ffi::Identifier invalid_id; - invalid_id.kind = "invalid"; - invalid_id.length = 0; + iggy::ffi::Consumer consumer{}; + consumer.kind = iggy::ffi::ConsumerKind::Consumer; + consumer.id.kind = "invalid"; + consumer.id.length = 0; - ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer", - invalid_id, "offset", 0, 10, false), + ASSERT_THROW(client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + std::move(consumer), "offset", 0, 10, false), std::exception); } @@ -1527,7 +1534,7 @@ TEST_F(LowLevelE2E_Message, ConsumerGroupCreateJoinAndPollMessages) { partition_id_bytes(0), std::move(messages)); auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, - "consumer_group", make_numeric_identifier(group.id), "offset", 0, 100, false); + iggy::Consumer::Group(group.id), "offset", 0, 100, false); ASSERT_EQ(polled.count, 10u); ASSERT_EQ(polled.messages.size(), 10u); @@ -1544,3 +1551,119 @@ TEST_F(LowLevelE2E_Message, ConsumerGroupCreateJoinAndPollMessages) { make_numeric_identifier(group.id)); ASSERT_EQ(group_after_leave.members_count, 0u); } + +TEST_F(LowLevelE2E_Message, PollMessagesWithDistinctConsumersKeepsOffsetsIndependent) { + RecordProperty("description", "Each named consumer owns its offset, so both read the whole partition."); + const std::string stream_name = GetRandomName(); + iggy::ffi::Client *client = GetLoggedInClient(); + + client->create_stream(stream_name); + auto stream = client->get_stream(make_string_identifier(stream_name)); + TrackStream(stream.id); + const std::string topic_name = GetRandomName(); + client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", + {}); + + rust::Vec messages; + for (std::uint32_t i = 0; i < 3; i++) { + auto msg = + iggy::ffi::make_message(to_payload("isolated-" + std::to_string(i)), rust::Vec()); + messages.push_back(std::move(msg)); + } + client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", + partition_id_bytes(0), std::move(messages)); + + for (const std::string &consumer_name : + {std::string("isolation-consumer-a"), std::string("isolation-consumer-b")}) { + auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single(consumer_name), "next", 0, 10, true); + + ASSERT_EQ(polled.count, 3u) << "Consumer " << consumer_name << " did not read the whole partition"; + for (std::uint32_t i = 0; i < 3; i++) { + std::string expected = "isolated-" + std::to_string(i); + std::string actual(polled.messages[i].payload.begin(), polled.messages[i].payload.end()); + EXPECT_EQ(actual, expected) << "Payload mismatch at offset " << i; + } + } +} + +TEST_F(LowLevelE2E_Message, PollMessagesWithSharedConsumerSplitsThePartition) { + RecordProperty("description", "Two polls under one consumer name share a stored offset, so the second reads none."); + const std::string stream_name = GetRandomName(); + iggy::ffi::Client *client = GetLoggedInClient(); + + client->create_stream(stream_name); + auto stream = client->get_stream(make_string_identifier(stream_name)); + TrackStream(stream.id); + const std::string topic_name = GetRandomName(); + client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0, "server_default", + {}); + + rust::Vec messages; + for (std::uint32_t i = 0; i < 3; i++) { + auto msg = + iggy::ffi::make_message(to_payload("shared-" + std::to_string(i)), rust::Vec()); + messages.push_back(std::move(msg)); + } + client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", + partition_id_bytes(0), std::move(messages)); + + auto first_poll = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("shared-consumer"), "next", 0, 10, true); + ASSERT_EQ(first_poll.count, 3u); + + auto second_poll = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, + iggy::Consumer::Single("shared-consumer"), "next", 0, 10, true); + ASSERT_EQ(second_poll.count, 0u); +} + +TEST_F(LowLevelE2E_Message, PollMessagesWithConsumerGroupReadsAssignedPartitions) { + RecordProperty("description", + "A group member polling kAnyPartitionId reads its assigned partitions, one per call, rather than " + "falling back to partition 0."); + const std::string stream_name = GetRandomName(); + iggy::ffi::Client *client = GetLoggedInClient(); + + client->create_stream(stream_name); + auto stream = client->get_stream(make_string_identifier(stream_name)); + TrackStream(stream.id); + const std::string topic_name = GetRandomName(); + client->create_topic(make_numeric_identifier(stream.id), topic_name, 2, "none", "never_expire", 0, "server_default", + {}); + + const std::string group_name = GetRandomName(); + client->create_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0), group_name); + TrackConsumerGroup(stream_name, topic_name, group_name); + client->join_consumer_group(make_numeric_identifier(stream.id), make_numeric_identifier(0), + make_string_identifier(group_name)); + + for (std::uint32_t partition_id = 0; partition_id < 2; partition_id++) { + rust::Vec messages; + for (std::uint32_t i = 0; i < 2; i++) { + auto msg = iggy::ffi::make_message(to_payload("p" + std::to_string(partition_id) + "-" + std::to_string(i)), + rust::Vec()); + messages.push_back(std::move(msg)); + } + client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id", + partition_id_bytes(partition_id), std::move(messages)); + } + + // Each poll takes the next partition the member owns, so two polls cover + // both. A client falling back to partition 0 would read it twice instead. + std::unordered_set polled_partitions; + std::unordered_set polled_payloads; + for (std::uint32_t poll = 0; poll < 2; poll++) { + auto polled = + client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), iggy::kAnyPartitionId, + iggy::Consumer::Group(group_name), "next", 0, 10, true); + + ASSERT_EQ(polled.count, 2u) << "Poll " << poll << " did not read a whole partition"; + polled_partitions.insert(polled.partition_id); + for (const auto &message : polled.messages) { + polled_payloads.insert(std::string(message.payload.begin(), message.payload.end())); + } + } + + EXPECT_EQ(polled_partitions, (std::unordered_set{0, 1})); + EXPECT_EQ(polled_payloads, (std::unordered_set{"p0-0", "p0-1", "p1-0", "p1-1"})); +} diff --git a/foreign/cpp/tests/e2e/topic.cpp b/foreign/cpp/tests/e2e/topic.cpp index 5bc3bdaa4e..b6894c0e6a 100644 --- a/foreign/cpp/tests/e2e/topic.cpp +++ b/foreign/cpp/tests/e2e/topic.cpp @@ -1048,8 +1048,8 @@ TEST_F(LowLevelE2E_Topic, UpdateTopicDoesNotChangeMessages) { ASSERT_NO_THROW({ const auto polled = client->poll_messages(make_numeric_identifier(created_stream.id), - make_string_identifier(updated_topic_name), 0, "consumer", - make_numeric_identifier(1), "offset", 0, 10, false); + make_string_identifier(updated_topic_name), 0, + iggy::Consumer::Single(1), "offset", 0, 10, false); ASSERT_EQ(polled.count, 1u); ASSERT_EQ(polled.messages.size(), 1u); const std::string actual(polled.messages[0].payload.begin(), polled.messages[0].payload.end()); diff --git a/foreign/cpp/tests/unit/consumer.cpp b/foreign/cpp/tests/unit/consumer.cpp new file mode 100644 index 0000000000..fda147b5d0 --- /dev/null +++ b/foreign/cpp/tests/unit/consumer.cpp @@ -0,0 +1,82 @@ +/* + * 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. + */ + +#include +#include +#include + +#include + +#include "iggy.hpp" + +namespace { + +std::string identifier_text(const iggy::ffi::Identifier &identifier) { + return std::string(identifier.value.begin(), identifier.value.end()); +} + +} // namespace + +TEST(ConsumerTest, SingleFromNameCarriesConsumerKind) { + const auto consumer = iggy::Consumer::Single("order-processor"); + + EXPECT_EQ(consumer.kind, iggy::ffi::ConsumerKind::Consumer); + EXPECT_EQ(consumer.id.kind, "string"); + EXPECT_EQ(identifier_text(consumer.id), "order-processor"); +} + +TEST(ConsumerTest, SingleFromNumberCarriesConsumerKind) { + const auto consumer = iggy::Consumer::Single(7); + + EXPECT_EQ(consumer.kind, iggy::ffi::ConsumerKind::Consumer); + EXPECT_EQ(consumer.id.kind, "numeric"); + EXPECT_EQ(consumer.id.length, 4u); +} + +TEST(ConsumerTest, GroupFromNameCarriesConsumerGroupKind) { + const auto consumer = iggy::Consumer::Group("order-processors"); + + EXPECT_EQ(consumer.kind, iggy::ffi::ConsumerKind::ConsumerGroup); + EXPECT_EQ(consumer.id.kind, "string"); + EXPECT_EQ(identifier_text(consumer.id), "order-processors"); +} + +TEST(ConsumerTest, GroupFromNumberCarriesConsumerGroupKind) { + const auto consumer = iggy::Consumer::Group(7); + + EXPECT_EQ(consumer.kind, iggy::ffi::ConsumerKind::ConsumerGroup); + EXPECT_EQ(consumer.id.kind, "numeric"); + EXPECT_EQ(consumer.id.length, 4u); +} + +TEST(ConsumerTest, RejectsEmptyName) { + EXPECT_THROW(iggy::Consumer::Single(""), iggy::IggyException); + EXPECT_THROW(iggy::Consumer::Group(""), iggy::IggyException); +} + +TEST(ConsumerTest, RejectsNameLongerThan255Bytes) { + const std::string too_long_name(256, 'a'); + + EXPECT_THROW(iggy::Consumer::Single(too_long_name), iggy::IggyException); + EXPECT_THROW(iggy::Consumer::Group(too_long_name), iggy::IggyException); +} + +TEST(AnyPartitionIdTest, LeavesThePartitionToTheServer) { + EXPECT_EQ(iggy::kAnyPartitionId, std::numeric_limits::max()); +}