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
The following commit(s) were added to refs/heads/cpp-high-level-client-2 by
this push:
new 4d7af5d34 fixes
4d7af5d34 is described below
commit 4d7af5d34bca4647e678891112e16e8f90a03414
Author: Rimuksh Kansal <[email protected]>
AuthorDate: Mon Sep 21 16:02:54 2026 +0900
fixes
---
.../features/step_definitions/background_steps.cpp | 3 +--
foreign/cpp/include/iggy.hpp | 29 +++++++++++++-------
foreign/cpp/src/client.cpp | 31 ++++++++++++++++------
foreign/cpp/tests/e2e/consumer_group.cpp | 12 ++++-----
foreign/cpp/tests/unit/unit_tests.cpp | 27 +++++++++++++++++++
5 files changed, 76 insertions(+), 26 deletions(-)
diff --git a/bdd/cpp/features/step_definitions/background_steps.cpp
b/bdd/cpp/features/step_definitions/background_steps.cpp
index eaa0e7b9e..33fcb008e 100644
--- a/bdd/cpp/features/step_definitions/background_steps.cpp
+++ b/bdd/cpp/features/step_definitions/background_steps.cpp
@@ -37,8 +37,7 @@ namespace {
std::string required_env(const char *name) {
const char *value = std::getenv(name);
if (value == nullptr || *value == '\0') {
- throw std::runtime_error(std::string(name) +
- " must be set; run the suite via
scripts/run-bdd-tests.sh");
+ throw std::runtime_error(std::string(name) + " must be set; run the
suite via scripts/run-bdd-tests.sh");
}
return std::string(value);
}
diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp
index eb93ed677..c213638a8 100644
--- a/foreign/cpp/include/iggy.hpp
+++ b/foreign/cpp/include/iggy.hpp
@@ -1137,8 +1137,8 @@ class ConsumerGroupMember final {
[[nodiscard]] std::uint32_t Id() const noexcept { return id_; }
/**
- * @brief Returns the number of partitions assigned to this member.
- * @return Number of entries in Partitions().
+ * @brief Returns the server-reported number of partitions assigned to
this member.
+ * @return Partition count reported by the server.
*/
[[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return
partitions_count_; }
@@ -1237,8 +1237,8 @@ class ConsumerGroupDetails final {
[[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return
partitions_count_; }
/**
- * @brief Returns the number of members in the group.
- * @return Number of entries in Members().
+ * @brief Returns the server-reported number of members in the group.
+ * @return Member count reported by the server.
*/
[[nodiscard]] std::uint32_t MembersCount() const noexcept { return
members_count_; }
@@ -2549,7 +2549,10 @@ class IggyBlockingClient final {
* @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.
+ * @param partition_id Partition whose offset is stored, or `std::nullopt`
+ * to omit the partition from the request. The maximum
+ * `std::uint32_t` value is rejected because it is reserved by the
+ * FFI representation.
* @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.
@@ -2558,7 +2561,7 @@ class IggyBlockingClient final {
const Identifier &stream,
const Identifier &topic,
std::uint64_t offset,
- std::uint32_t partition_id);
+ std::optional<std::uint32_t> partition_id =
std::nullopt);
/**
* @brief Retrieves the stored offset for a consumer or consumer group.
@@ -2574,7 +2577,10 @@ class IggyBlockingClient final {
* @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.
+ * @param partition_id Partition whose offset is retrieved, or
+ * `std::nullopt` to omit the partition from the request. The
+ * maximum `std::uint32_t` value is rejected because it is reserved
+ * by the FFI representation.
* @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
@@ -2584,7 +2590,7 @@ class IggyBlockingClient final {
ConsumerOffsetInfo GetConsumerOffset(const Consumer &consumer,
const Identifier &stream,
const Identifier &topic,
- std::uint32_t partition_id);
+ std::optional<std::uint32_t>
partition_id = std::nullopt);
/**
* @brief Deletes the stored offset for a consumer or consumer group.
@@ -2601,7 +2607,10 @@ class IggyBlockingClient final {
* @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.
+ * @param partition_id Partition whose offset is deleted, or `std::nullopt`
+ * to omit the partition from the request. The maximum
+ * `std::uint32_t` value is rejected because it is reserved by the
+ * FFI representation.
* @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
@@ -2610,7 +2619,7 @@ class IggyBlockingClient final {
void DeleteConsumerOffset(const Consumer &consumer,
const Identifier &stream,
const Identifier &topic,
- std::uint32_t partition_id);
+ std::optional<std::uint32_t> partition_id =
std::nullopt);
private:
explicit IggyBlockingClient(ffi::Client *client);
diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp
index afe1b551c..b61b2fd2b 100644
--- a/foreign/cpp/src/client.cpp
+++ b/foreign/cpp/src/client.cpp
@@ -307,30 +307,45 @@ void IggyBlockingClient::StoreConsumerOffset(const
Consumer &consumer,
const Identifier &stream,
const Identifier &topic,
const std::uint64_t offset,
- const std::uint32_t partition_id)
{
+ const
std::optional<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);
+ constexpr auto unspecified_partition_id =
std::numeric_limits<std::uint32_t>::max();
+ if (partition_id == unspecified_partition_id) {
+ throw std::invalid_argument("partition_id cannot be the maximum
std::uint32_t value");
+ }
+ const auto ffi_partition_id =
partition_id.value_or(unspecified_partition_id);
+ Handle()->store_consumer_offset(stream.ToFfi(), topic.ToFfi(),
ffi_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) {
+ const
std::optional<std::uint32_t> partition_id) {
return RethrowAsIggyException([this, &consumer, &stream, &topic,
partition_id] {
+ constexpr auto unspecified_partition_id =
std::numeric_limits<std::uint32_t>::max();
+ if (partition_id == unspecified_partition_id) {
+ throw std::invalid_argument("partition_id cannot be the maximum
std::uint32_t value");
+ }
+ const auto ffi_partition_id =
partition_id.value_or(unspecified_partition_id);
return ConsumerOffsetInfo::FromFfi(Handle()->get_consumer_offset(
- stream.ToFfi(), topic.ToFfi(), partition_id,
std::string(consumer.KindName()), consumer.Id().ToFfi()));
+ stream.ToFfi(), topic.ToFfi(), ffi_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) {
+ const
std::optional<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());
+ constexpr auto unspecified_partition_id =
std::numeric_limits<std::uint32_t>::max();
+ if (partition_id == unspecified_partition_id) {
+ throw std::invalid_argument("partition_id cannot be the maximum
std::uint32_t value");
+ }
+ const auto ffi_partition_id =
partition_id.value_or(unspecified_partition_id);
+ Handle()->delete_consumer_offset(stream.ToFfi(), topic.ToFfi(),
ffi_partition_id,
+ std::string(consumer.KindName()),
consumer.Id().ToFfi());
});
}
diff --git a/foreign/cpp/tests/e2e/consumer_group.cpp
b/foreign/cpp/tests/e2e/consumer_group.cpp
index 79b5730b7..87ca3f29a 100644
--- a/foreign/cpp/tests/e2e/consumer_group.cpp
+++ b/foreign/cpp/tests/e2e/consumer_group.cpp
@@ -1156,7 +1156,7 @@ TEST_F(E2E_ConsumerGroup,
DeleteConsumerGroupAndRecreateWithSameNameSucceeds) {
}
TEST_F(E2E_ConsumerGroup, StoreGetAndDeleteConsumerOffsetSucceeds) {
- RecordProperty("description", "Stores, retrieves, and deletes an
individual consumer offset.");
+ RecordProperty("description", "Retrieves a partition-0 offset with no
partition specified.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
auto client = GetLoggedInHighLevelClient();
@@ -1177,17 +1177,17 @@ TEST_F(E2E_ConsumerGroup,
StoreGetAndDeleteConsumerOffsetSucceeds) {
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);
+ const auto offset =
+ client.GetConsumerOffset(consumer,
iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
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);
+ ASSERT_THROW(
+ client.GetConsumerOffset(consumer,
iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
}
TEST_F(E2E_ConsumerGroup, StoreConsumerOffsetOnEmptyPartitionThrows) {
diff --git a/foreign/cpp/tests/unit/unit_tests.cpp
b/foreign/cpp/tests/unit/unit_tests.cpp
index b1b0be60d..410e91fbd 100644
--- a/foreign/cpp/tests/unit/unit_tests.cpp
+++ b/foreign/cpp/tests/unit/unit_tests.cpp
@@ -352,6 +352,7 @@ TEST(IggyBlockingClientTest, MovedFromOperationsThrow) {
const auto stream = iggy::Identifier::String("stream");
const auto topic = iggy::Identifier::String("topic");
+ const auto group = iggy::Identifier::String("group");
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.
@@ -375,11 +376,37 @@ 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.CreateConsumerGroup(stream, topic, "group"),
iggy::IggyException);
+ EXPECT_THROW(client.GetConsumerGroup(stream, topic, group),
iggy::IggyException);
+ EXPECT_THROW(client.GetConsumerGroups(stream, topic), iggy::IggyException);
+ EXPECT_THROW(client.DeleteConsumerGroup(stream, topic, group),
iggy::IggyException);
+ EXPECT_THROW(client.JoinConsumerGroup(stream, topic, group),
iggy::IggyException);
+ EXPECT_THROW(client.LeaveConsumerGroup(stream, topic, group),
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(IggyBlockingClientTest, ConsumerOffsetOperationsRejectMaximumPartitionId)
{
+ auto client = iggy::IggyBlockingClient::Builder().Build();
+ const auto stream = iggy::Identifier::String("stream");
+ const auto topic = iggy::Identifier::String("topic");
+ const auto consumer =
iggy::Consumer::Single(iggy::Identifier::Numeric(1));
+ const auto maximum_partition = std::numeric_limits<std::uint32_t>::max();
+ const auto expect_rejection = [](auto &&operation) {
+ try {
+ operation();
+ FAIL() << "Expected the maximum std::uint32_t partition_id to be
rejected";
+ } catch (const iggy::IggyException &error) {
+ EXPECT_STREQ(error.what(), "partition_id cannot be the maximum
std::uint32_t value");
+ }
+ };
+
+ expect_rejection([&] { client.StoreConsumerOffset(consumer, stream, topic,
0, maximum_partition); });
+ expect_rejection([&] { (void)client.GetConsumerOffset(consumer, stream,
topic, maximum_partition); });
+ expect_rejection([&] { client.DeleteConsumerOffset(consumer, stream,
topic, maximum_partition); });
+}
+
TEST(AutoLoginKindTest, HasStableDiscriminantsAndZeroInitializedDefault) {
EXPECT_EQ(static_cast<std::uint8_t>(iggy::ffi::AutoLoginKind::Disabled),
0u);
EXPECT_EQ(static_cast<std::uint8_t>(iggy::ffi::AutoLoginKind::UsernamePassword),
1u);
