This is an automated email from the ASF dual-hosted git repository. slbotbm pushed a commit to branch cpp-high-level-client-2 in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 74a4ee6e1b9119153eac74aabe35895fe57f9529 Author: Rimuksh Kansal <[email protected]> AuthorDate: Mon Sep 14 21:20:10 2026 +0900 add offset management functions --- foreign/cpp/include/iggy.hpp | 122 ++++ foreign/cpp/src/client.cpp | 31 + foreign/cpp/src/type_conversions.cpp | 4 + foreign/cpp/tests/e2e/client.cpp | 2 - foreign/cpp/tests/e2e/consumer_group.cpp | 1049 ++++++++++++++++++++++++++++++ foreign/cpp/tests/unit/unit_tests.cpp | 8 +- 6 files changed, 1212 insertions(+), 4 deletions(-) diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp index 7b025a2d6..127853d9d 100644 --- a/foreign/cpp/include/iggy.hpp +++ b/foreign/cpp/include/iggy.hpp @@ -41,6 +41,8 @@ namespace iggy { +class Consumer; +class ConsumerOffsetInfo; class IggyBlockingClient; class LoginInfo; class Partition; @@ -178,6 +180,73 @@ class Identifier final { std::variant<std::uint32_t, std::string> value_; }; +/** + * @brief Identifies a single consumer or consumer group for offset operations. + */ +class Consumer final { + public: + enum class Kind { Single, Group }; + + /** + * @brief Identifies an individual consumer. + * @param id Consumer ID or name. + * @return Individual consumer identity. + */ + static Consumer Single(Identifier id) { return Consumer(Kind::Single, std::move(id)); } + + /** + * @brief Identifies a consumer group. + * @param id Consumer group ID or name. + * @return Consumer group identity. + */ + static Consumer Group(Identifier id) { return Consumer(Kind::Group, std::move(id)); } + + /** @brief Returns whether this identity represents a single consumer or a group. */ + [[nodiscard]] Kind Type() const noexcept { return kind_; } + + /** @brief Returns the consumer or consumer group identifier. */ + [[nodiscard]] const Identifier &Id() const noexcept { return id_; } + + private: + Consumer(Kind kind, Identifier id) : kind_(kind), id_(std::move(id)) {} + + [[nodiscard]] std::string_view KindName() const noexcept { + return kind_ == Kind::Single ? "consumer" : "consumer_group"; + } + + friend class IggyBlockingClient; + + Kind kind_; + Identifier id_; +}; + +/** + * @brief Consumer offset state returned by GetConsumerOffset(). + */ +class ConsumerOffsetInfo final { + public: + /** @brief Returns the partition associated with the stored offset. */ + [[nodiscard]] std::uint32_t PartitionId() const noexcept { return partition_id_; } + + /** @brief Returns the partition's current message offset. */ + [[nodiscard]] std::uint64_t CurrentOffset() const noexcept { return current_offset_; } + + /** @brief Returns the offset stored for the consumer identity. */ + [[nodiscard]] std::uint64_t StoredOffset() const noexcept { return stored_offset_; } + + private: + ConsumerOffsetInfo(std::uint32_t partition_id, std::uint64_t current_offset, std::uint64_t stored_offset) + : partition_id_(partition_id), current_offset_(current_offset), stored_offset_(stored_offset) {} + + static ConsumerOffsetInfo FromFfi(ffi::ConsumerOffsetInfo offset); + + friend class IggyBlockingClient; + + std::uint32_t partition_id_; + std::uint64_t current_offset_; + std::uint64_t stored_offset_; +}; + /** * @brief Type tag for a HeaderField payload. * @@ -2156,6 +2225,59 @@ class IggyBlockingClient final { */ void LeaveConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group); + /** + * @brief Stores an offset for a consumer or consumer group. + * + * @param consumer Consumer identity that owns the offset. + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param offset Message offset to store. + * @param partition_id Partition whose offset is stored. + * @throws IggyException if an identifier, partition, or offset is invalid; + * the resource does not exist; the client is unauthenticated; the + * caller lacks permission; or the request fails. + */ + void StoreConsumerOffset(const Consumer &consumer, + const Identifier &stream, + const Identifier &topic, + std::uint64_t offset, + std::uint32_t partition_id); + + /** + * @brief Retrieves the stored offset for a consumer or consumer group. + * + * @param consumer Consumer identity that owns the offset. + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param partition_id Partition whose offset is retrieved. + * @return Partition state and the stored consumer offset. + * @throws IggyException if an identifier or partition is invalid; the + * resource or stored offset does not exist; the client is + * unauthenticated; the caller lacks permission; or the request + * fails. + */ + ConsumerOffsetInfo GetConsumerOffset(const Consumer &consumer, + const Identifier &stream, + const Identifier &topic, + std::uint32_t partition_id); + + /** + * @brief Deletes the stored offset for a consumer or consumer group. + * + * @param consumer Consumer identity that owns the offset. + * @param stream Parent stream, addressed by numeric ID or name. + * @param topic Parent topic, addressed by numeric ID or name. + * @param partition_id Partition whose offset is deleted. + * @throws IggyException if an identifier or partition is invalid; the + * resource or stored offset does not exist; the client is + * unauthenticated; the caller lacks permission; or the request + * fails. + */ + void DeleteConsumerOffset(const Consumer &consumer, + const Identifier &stream, + const Identifier &topic, + std::uint32_t partition_id); + private: explicit IggyBlockingClient(ffi::Client *client); diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp index cb5041945..afe1b551c 100644 --- a/foreign/cpp/src/client.cpp +++ b/foreign/cpp/src/client.cpp @@ -303,6 +303,37 @@ void IggyBlockingClient::LeaveConsumerGroup(const Identifier &stream, }); } +void IggyBlockingClient::StoreConsumerOffset(const Consumer &consumer, + const Identifier &stream, + const Identifier &topic, + const std::uint64_t offset, + const std::uint32_t partition_id) { + RethrowAsIggyException([this, &consumer, &stream, &topic, offset, partition_id] { + Handle()->store_consumer_offset(stream.ToFfi(), topic.ToFfi(), partition_id, std::string(consumer.KindName()), + consumer.Id().ToFfi(), offset); + }); +} + +ConsumerOffsetInfo IggyBlockingClient::GetConsumerOffset(const Consumer &consumer, + const Identifier &stream, + const Identifier &topic, + const std::uint32_t partition_id) { + return RethrowAsIggyException([this, &consumer, &stream, &topic, partition_id] { + return ConsumerOffsetInfo::FromFfi(Handle()->get_consumer_offset( + stream.ToFfi(), topic.ToFfi(), partition_id, std::string(consumer.KindName()), consumer.Id().ToFfi())); + }); +} + +void IggyBlockingClient::DeleteConsumerOffset(const Consumer &consumer, + const Identifier &stream, + const Identifier &topic, + const std::uint32_t partition_id) { + RethrowAsIggyException([this, &consumer, &stream, &topic, partition_id] { + Handle()->delete_consumer_offset(stream.ToFfi(), topic.ToFfi(), partition_id, std::string(consumer.KindName()), + consumer.Id().ToFfi()); + }); +} + IggyBlockingClient::IggyBlockingClient(ffi::Client *client) : client_(client) { if (client_ == nullptr) { throw IggyException("Could not create Iggy client"); diff --git a/foreign/cpp/src/type_conversions.cpp b/foreign/cpp/src/type_conversions.cpp index ce8251050..07068d7bd 100644 --- a/foreign/cpp/src/type_conversions.cpp +++ b/foreign/cpp/src/type_conversions.cpp @@ -42,6 +42,10 @@ ffi::Identifier Identifier::ToFfi() const { return identifier; } +ConsumerOffsetInfo ConsumerOffsetInfo::FromFfi(ffi::ConsumerOffsetInfo offset) { + return ConsumerOffsetInfo(offset.partition_id, offset.current_offset, offset.stored_offset); +} + HeaderField HeaderField::FromFfi(ffi::HeaderField field) { return HeaderField(static_cast<HeaderKind>(field.kind), std::vector<std::uint8_t>(field.value.begin(), field.value.end())); diff --git a/foreign/cpp/tests/e2e/client.cpp b/foreign/cpp/tests/e2e/client.cpp index 38ae37910..b2b25f027 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 <algorithm> #include <chrono> diff --git a/foreign/cpp/tests/e2e/consumer_group.cpp b/foreign/cpp/tests/e2e/consumer_group.cpp index cc8d6740d..79b5730b7 100644 --- a/foreign/cpp/tests/e2e/consumer_group.cpp +++ b/foreign/cpp/tests/e2e/consumer_group.cpp @@ -1154,3 +1154,1052 @@ TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupAndRecreateWithSameNameSucceeds) { ASSERT_TRUE(recreated_group.Members().empty()); }); } + +TEST_F(E2E_ConsumerGroup, StoreGetAndDeleteConsumerOffsetSucceeds) { + RecordProperty("description", "Stores, retrieves, and deletes an individual consumer offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + + const auto offset = client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0); + EXPECT_EQ(offset.PartitionId(), 0u); + EXPECT_EQ(offset.CurrentOffset(), 0u); + EXPECT_EQ(offset.StoredOffset(), 0u); + + ASSERT_NO_THROW(client.DeleteConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetOnEmptyPartitionThrows) { + RecordProperty("description", "Rejects offsets for a partition that has not issued any message offsets."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + for (const std::uint64_t offset : {0u, 1u}) { + SCOPED_TRACE(offset); + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), offset, 0), + iggy::IggyException); + } +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetAcceptsOffsetsAtValidBounds) { + RecordProperty("description", "Stores offsets below and at the partition's current offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 5; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 2, 0)); + EXPECT_EQ( + client + .GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 2u); + + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 4, 0)); + const auto current = client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0); + EXPECT_EQ(current.CurrentOffset(), 4u); + EXPECT_EQ(current.StoredOffset(), 4u); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetPastCurrentOffsetThrowsWithoutChangingStoredOffset) { + RecordProperty("description", "Rejects an offset past the partition head without replacing the stored offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 5; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 2, 0)); + + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 5, 0), + iggy::IggyException); + EXPECT_EQ( + client + .GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 2u); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetBeforeLoginThrows) { + RecordProperty("description", "Rejects storing an offset before login and after disconnect."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto setup_client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + auto client = GetLoggedOutHighLevelClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 1; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0), + iggy::IggyException); + ASSERT_NO_THROW(client.Connect()); + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0), + iggy::IggyException); + ASSERT_NO_THROW(client.Login("iggy", "iggy")); + ASSERT_NO_THROW(client.Disconnect()); + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetOnNonExistentResourcesThrows) { + RecordProperty("description", "Rejects missing streams, topics, and partitions."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string missing_stream_name = GetRandomName(); + const std::string missing_topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 1; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(missing_stream_name), + iggy::Identifier::String(topic_name), 0, 0), + iggy::IggyException); + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(missing_topic_name), 0, 0), + iggy::IggyException); + ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 1), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetUpdatesExistingOffset) { + RecordProperty("description", "Replaces a previously stored offset for the same consumer and partition."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 4; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1, 0)); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 3, 0)); + + EXPECT_EQ( + client + .GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 3u); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetKeepsConsumerOffsetsIndependent) { + RecordProperty("description", "Stores independent offsets for different consumers on the same partition."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto first_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + const auto second_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(2)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 3; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + ASSERT_NO_THROW(client.StoreConsumerOffset(second_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 2, 0)); + + EXPECT_EQ(client + .GetConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 0u); + EXPECT_EQ(client + .GetConsumerOffset(second_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 2u); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetForOwnedConsumerGroupPartitionSucceeds) { + RecordProperty("description", "Stores an offset for a partition owned by the current consumer group member."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string group_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 1; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); + TrackConsumerGroup(stream_name, topic_name, group_name); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + const auto consumer_group = iggy::Consumer::Group(iggy::Identifier::String(group_name)); + + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + EXPECT_EQ(client + .GetConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 0u); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerGroupOffsetForUnownedPartitionThrows) { + RecordProperty("description", "Rejects storing a group offset when the current client does not own the partition."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string group_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 1; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); + TrackConsumerGroup(stream_name, topic_name, group_name); + const auto consumer_group = iggy::Consumer::Group(iggy::Identifier::String(group_name)); + + ASSERT_THROW(client.StoreConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0), + iggy::IggyException); + ASSERT_THROW(client.GetConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetForNonExistentConsumerGroupThrows) { + RecordProperty("description", "Rejects named and numeric consumer groups that do not exist."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 1; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + const auto missing_by_name = iggy::Consumer::Group(iggy::Identifier::String(GetRandomName())); + const auto missing_by_id = iggy::Consumer::Group(iggy::Identifier::Numeric(999'999)); + for (const auto *consumer_group : {&missing_by_name, &missing_by_id}) { + ASSERT_THROW(client.StoreConsumerOffset(*consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0), + iggy::IggyException); + } +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetSupportsNamedAndNumericIdentifiers) { + RecordProperty("description", "Stores offsets using named and numeric stream, topic, and consumer identifiers."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + const auto stream = client.CreateStream(stream_name); + TrackStream(stream_name); + const auto topic = client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1)); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 2; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + const auto named_consumer = iggy::Consumer::Single(iggy::Identifier::String(GetRandomName())); + const auto numeric_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(42)); + ASSERT_NO_THROW(client.StoreConsumerOffset(named_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + ASSERT_NO_THROW(client.StoreConsumerOffset(numeric_consumer, iggy::Identifier::Numeric(stream.Id()), + iggy::Identifier::Numeric(topic.Id()), 1, 0)); + + EXPECT_EQ(client + .GetConsumerOffset(named_consumer, iggy::Identifier::Numeric(stream.Id()), + iggy::Identifier::Numeric(topic.Id()), 0) + .StoredOffset(), + 0u); + EXPECT_EQ(client + .GetConsumerOffset(numeric_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 1u); +} + +TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetWithoutPermissionThrowsWithoutChangingOffset) { + RecordProperty("description", "Rejects an unauthorized offset write without changing the existing value."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string username = GetRandomName(50); + const std::string password = "secret123"; + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + auto *user_admin_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 3; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1, 0)); + ASSERT_NO_THROW(CreateUser(user_admin_client, username, password, iggy::ffi::UserStatus::Active, true, + iggy::ffi::Permissions{})); + auto restricted_client = GetLoggedInHighLevelClient(username, password); + + ASSERT_THROW(restricted_client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 2, 0), + iggy::IggyException); + EXPECT_EQ( + client + .GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 1u); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetReturnsAllFieldsForNonZeroPartition) { + RecordProperty("description", "Returns the requested partition, its current offset, and the stored offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(2))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 5; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(1), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 2, 1)); + + const auto offset = client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1); + EXPECT_EQ(offset.PartitionId(), 1u); + EXPECT_EQ(offset.CurrentOffset(), 4u); + EXPECT_EQ(offset.StoredOffset(), 2u); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetWithoutStoredOffsetThrows) { + RecordProperty("description", "Rejects retrieving an offset that has not been stored for the consumer."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetBeforeLoginThrows) { + RecordProperty("description", "Rejects retrieving an offset before login and after disconnect."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto setup_client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + auto client = GetLoggedOutHighLevelClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(setup_client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_NO_THROW(client.Connect()); + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_NO_THROW(client.Login("iggy", "iggy")); + ASSERT_NO_THROW({ + const auto offset = client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0); + EXPECT_EQ(offset.StoredOffset(), 0u); + }); + ASSERT_NO_THROW(client.Disconnect()); + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetOnNonExistentResourcesThrows) { + RecordProperty("description", "Rejects missing streams, topics, and partitions when retrieving an offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string missing_stream_name = GetRandomName(); + const std::string missing_topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(missing_stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(missing_topic_name), 0), + iggy::IggyException); + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetWithoutPermissionThrows) { + RecordProperty("description", "Rejects retrieving an existing offset without poll permission."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string username = GetRandomName(50); + const std::string password = "secret123"; + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + auto *user_admin_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + ASSERT_NO_THROW(CreateUser(user_admin_client, username, password, iggy::ffi::UserStatus::Active, true, + iggy::ffi::Permissions{})); + auto restricted_client = GetLoggedInHighLevelClient(username, password); + + ASSERT_THROW(restricted_client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerGroupOffsetCanBeReadByNonMember) { + RecordProperty("description", "Allows an authenticated non-member to retrieve an existing consumer group offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string group_name = GetRandomName(); + auto owner_client = GetLoggedInHighLevelClient(); + auto reader_client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(owner_client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(owner_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 3; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + const auto group = owner_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name); + TrackConsumerGroup(stream_name, topic_name, group_name); + ASSERT_NO_THROW(owner_client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + const auto named_group = iggy::Consumer::Group(iggy::Identifier::String(group_name)); + ASSERT_NO_THROW(owner_client.StoreConsumerOffset(named_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1, 0)); + + const auto numeric_group = iggy::Consumer::Group(iggy::Identifier::Numeric(group.Id())); + const auto offset = reader_client.GetConsumerOffset(numeric_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0); + EXPECT_EQ(offset.PartitionId(), 0u); + EXPECT_EQ(offset.CurrentOffset(), 2u); + EXPECT_EQ(offset.StoredOffset(), 1u); +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetForNonExistentConsumerGroupThrows) { + RecordProperty("description", + "Rejects retrieving offsets for named and numeric consumer groups that do not exist."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + const auto missing_by_name = iggy::Consumer::Group(iggy::Identifier::String(GetRandomName())); + const auto missing_by_id = iggy::Consumer::Group(iggy::Identifier::Numeric(999'999)); + for (const auto *consumer_group : {&missing_by_name, &missing_by_id}) { + ASSERT_THROW(client.GetConsumerOffset(*consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + } +} + +TEST_F(E2E_ConsumerGroup, GetConsumerOffsetReflectsAutoCommittedPoll) { + RecordProperty("description", "Returns the offset created by polling messages with auto-commit enabled."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(77)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 5; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + iggy::ffi::PolledMessages polled{}; + ASSERT_NO_THROW(polled = message_client->poll_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), 0, "consumer", + make_numeric_identifier(77), "next", 0, 3, true)); + ASSERT_EQ(polled.count, 3u); + ASSERT_EQ(polled.messages.size(), 3u); + EXPECT_EQ(polled.messages.front().offset, 0u); + EXPECT_EQ(polled.messages.back().offset, 2u); + + const auto offset = client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0); + EXPECT_EQ(offset.PartitionId(), 0u); + EXPECT_EQ(offset.CurrentOffset(), 4u); + EXPECT_EQ(offset.StoredOffset(), 2u); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetForMissingOffsetThrows) { + RecordProperty("description", "Rejects deleting offsets that were never stored or were already deleted."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto missing_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + const auto stored_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(2)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + ASSERT_THROW(client.DeleteConsumerOffset(missing_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + + ASSERT_NO_THROW(client.StoreConsumerOffset(stored_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + ASSERT_NO_THROW(client.DeleteConsumerOffset(stored_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + ASSERT_THROW(client.DeleteConsumerOffset(stored_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetRemovesOnlyRequestedConsumerAndPartition) { + RecordProperty("description", "Deletes only the requested consumer and partition offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto first_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + const auto second_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(2)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(2))); + rust::Vec<iggy::ffi::IggyMessageToSend> first_partition_messages; + rust::Vec<iggy::ffi::IggyMessageToSend> second_partition_messages; + for (std::uint32_t index = 0; index < 4; ++index) { + first_partition_messages.push_back(iggy::ffi::make_message( + to_payload("first-partition-offset-test-" + std::to_string(index)), rust::Vec<iggy::ffi::HeaderEntry>{})); + second_partition_messages.push_back(iggy::ffi::make_message( + to_payload("second-partition-offset-test-" + std::to_string(index)), rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(first_partition_messages))); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(1), std::move(second_partition_messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1, 0)); + ASSERT_NO_THROW(client.StoreConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 2, 1)); + ASSERT_NO_THROW(client.StoreConsumerOffset(second_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 3, 0)); + + ASSERT_NO_THROW(client.DeleteConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + + ASSERT_THROW(client.GetConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + EXPECT_EQ(client + .GetConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1) + .StoredOffset(), + 2u); + EXPECT_EQ(client + .GetConsumerOffset(second_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 3u); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetBeforeLoginThrows) { + RecordProperty("description", "Rejects deleting an offset before login and after disconnect."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto setup_client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + auto client = GetLoggedOutHighLevelClient(); + const auto first_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + const auto second_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(2)); + + ASSERT_NO_THROW(setup_client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(setup_client.StoreConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + ASSERT_NO_THROW(setup_client.StoreConsumerOffset(second_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + + ASSERT_THROW(client.DeleteConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_NO_THROW(client.Connect()); + ASSERT_THROW(client.DeleteConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_NO_THROW(client.Login("iggy", "iggy")); + ASSERT_NO_THROW(client.DeleteConsumerOffset(first_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + ASSERT_NO_THROW(client.Disconnect()); + ASSERT_THROW(client.DeleteConsumerOffset(second_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetOnNonExistentResourcesThrows) { + RecordProperty("description", "Rejects missing streams, topics, and partitions when deleting an offset."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string missing_stream_name = GetRandomName(); + const std::string missing_topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + ASSERT_THROW(client.DeleteConsumerOffset(consumer, iggy::Identifier::String(missing_stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_THROW(client.DeleteConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(missing_topic_name), 0), + iggy::IggyException); + ASSERT_THROW(client.DeleteConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetWithoutPermissionThrowsWithoutRemovingOffset) { + RecordProperty("description", "Rejects an unauthorized offset deletion without removing the existing value."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string username = GetRandomName(50); + const std::string password = "secret123"; + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + auto *user_admin_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 3; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1, 0)); + ASSERT_NO_THROW(CreateUser(user_admin_client, username, password, iggy::ffi::UserStatus::Active, true, + iggy::ffi::Permissions{})); + auto restricted_client = GetLoggedInHighLevelClient(username, password); + + ASSERT_THROW(restricted_client.DeleteConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + EXPECT_EQ( + client + .GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 1u); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetForOwnedConsumerGroupPartitionSucceeds) { + RecordProperty("description", "Deletes an offset for a consumer group partition owned by the current client."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string group_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + messages.push_back(iggy::ffi::make_message(to_payload("offset-test"), rust::Vec<iggy::ffi::HeaderEntry>{})); + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); + TrackConsumerGroup(stream_name, topic_name, group_name); + ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + const auto consumer_group = iggy::Consumer::Group(iggy::Identifier::String(group_name)); + ASSERT_NO_THROW(client.StoreConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + + ASSERT_NO_THROW(client.DeleteConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + ASSERT_THROW(client.GetConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupOffsetForUnownedPartitionThrowsWithoutRemovingOffset) { + RecordProperty("description", "Rejects deleting a group offset from an unowned partition without removing it."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + const std::string group_name = GetRandomName(); + auto owner_client = GetLoggedInHighLevelClient(); + auto non_member_client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + ASSERT_NO_THROW(owner_client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(owner_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 3; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + ASSERT_NO_THROW(owner_client.CreateConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), group_name)); + TrackConsumerGroup(stream_name, topic_name, group_name); + ASSERT_NO_THROW(owner_client.JoinConsumerGroup(iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), + iggy::Identifier::String(group_name))); + const auto consumer_group = iggy::Consumer::Group(iggy::Identifier::String(group_name)); + ASSERT_NO_THROW(owner_client.StoreConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 1, 0)); + + ASSERT_THROW(non_member_client.DeleteConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + EXPECT_EQ(owner_client + .GetConsumerOffset(consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 1u); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetForNonExistentConsumerGroupThrows) { + RecordProperty("description", "Rejects deleting offsets for named and numeric consumer groups that do not exist."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + + const auto missing_by_name = iggy::Consumer::Group(iggy::Identifier::String(GetRandomName())); + const auto missing_by_id = iggy::Consumer::Group(iggy::Identifier::Numeric(999'999)); + for (const auto *consumer_group : {&missing_by_name, &missing_by_id}) { + ASSERT_THROW(client.DeleteConsumerOffset(*consumer_group, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + } +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetSupportsNamedAndNumericIdentifiers) { + RecordProperty("description", "Deletes offsets using named and numeric stream, topic, and consumer identifiers."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + + const auto stream = client.CreateStream(stream_name); + TrackStream(stream_name); + const auto topic = client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1)); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 2; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + const auto named_consumer = iggy::Consumer::Single(iggy::Identifier::String(GetRandomName())); + const auto numeric_consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(42)); + ASSERT_NO_THROW(client.StoreConsumerOffset(named_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0, 0)); + ASSERT_NO_THROW(client.StoreConsumerOffset(numeric_consumer, iggy::Identifier::Numeric(stream.Id()), + iggy::Identifier::Numeric(topic.Id()), 1, 0)); + + ASSERT_NO_THROW(client.DeleteConsumerOffset(named_consumer, iggy::Identifier::Numeric(stream.Id()), + iggy::Identifier::Numeric(topic.Id()), 0)); + ASSERT_NO_THROW(client.DeleteConsumerOffset(numeric_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + ASSERT_THROW(client.GetConsumerOffset(named_consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); + ASSERT_THROW(client.GetConsumerOffset(numeric_consumer, iggy::Identifier::Numeric(stream.Id()), + iggy::Identifier::Numeric(topic.Id()), 0), + iggy::IggyException); +} + +TEST_F(E2E_ConsumerGroup, DeleteConsumerOffsetRemovesAutoCommittedOffset) { + RecordProperty("description", "Deletes an offset created by polling messages with auto-commit enabled."); + const std::string stream_name = GetRandomName(); + const std::string topic_name = GetRandomName(); + auto client = GetLoggedInHighLevelClient(); + auto *message_client = GetLoggedInClient(); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(88)); + + ASSERT_NO_THROW(client.CreateStream(stream_name)); + TrackStream(stream_name); + ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name, + iggy::TopicCreateOptions().SetPartitionsCount(1))); + rust::Vec<iggy::ffi::IggyMessageToSend> messages; + for (std::uint32_t index = 0; index < 5; ++index) { + messages.push_back(iggy::ffi::make_message(to_payload("offset-test-" + std::to_string(index)), + rust::Vec<iggy::ffi::HeaderEntry>{})); + } + ASSERT_NO_THROW(message_client->send_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), "partition_id", + partition_id_bytes(0), std::move(messages))); + + iggy::ffi::PolledMessages polled{}; + ASSERT_NO_THROW(polled = message_client->poll_messages(make_string_identifier(stream_name), + make_string_identifier(topic_name), 0, "consumer", + make_numeric_identifier(88), "next", 0, 3, true)); + ASSERT_EQ(polled.count, 3u); + EXPECT_EQ( + client + .GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name), 0) + .StoredOffset(), + 2u); + + ASSERT_NO_THROW(client.DeleteConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0)); + ASSERT_THROW(client.GetConsumerOffset(consumer, iggy::Identifier::String(stream_name), + iggy::Identifier::String(topic_name), 0), + iggy::IggyException); +} diff --git a/foreign/cpp/tests/unit/unit_tests.cpp b/foreign/cpp/tests/unit/unit_tests.cpp index c92831ffb..b1b0be60d 100644 --- a/foreign/cpp/tests/unit/unit_tests.cpp +++ b/foreign/cpp/tests/unit/unit_tests.cpp @@ -350,8 +350,9 @@ TEST(IggyBlockingClientTest, MovedFromOperationsThrow) { auto moved_to = std::move(client); (void)moved_to; - const auto stream = iggy::Identifier::String("stream"); - const auto topic = iggy::Identifier::String("topic"); + const auto stream = iggy::Identifier::String("stream"); + const auto topic = iggy::Identifier::String("topic"); + const auto consumer = iggy::Consumer::Single(iggy::Identifier::Numeric(1)); // Exercising the moved-from guard requires invoking every operation on the valid but empty source object. EXPECT_THROW(client.Connect(), iggy::IggyException); @@ -374,6 +375,9 @@ TEST(IggyBlockingClientTest, MovedFromOperationsThrow) { EXPECT_THROW(client.PurgeTopic(stream, topic), iggy::IggyException); EXPECT_THROW(client.CreatePartitions(stream, topic, 1), iggy::IggyException); EXPECT_THROW(client.DeletePartitions(stream, topic, 1), iggy::IggyException); + EXPECT_THROW(client.StoreConsumerOffset(consumer, stream, topic, 0, 0), iggy::IggyException); + EXPECT_THROW(client.GetConsumerOffset(consumer, stream, topic, 0), iggy::IggyException); + EXPECT_THROW(client.DeleteConsumerOffset(consumer, stream, topic, 0), iggy::IggyException); } TEST(AutoLoginKindTest, HasStableDiscriminantsAndZeroInitializedDefault) {
