You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/21 18:54:26 UTC
(iggy) branch master updated: feat(cpp): add functions related to consumers and offsets to high level client (#4180)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new da4f3188d feat(cpp): add functions related to consumers and offsets to high level client (#4180)
da4f3188d is described below
commit da4f3188d84f2c619177188f0958cc7c7523c141
Author: Rimuksh Kansal <ni...@gmail.com>
AuthorDate: Tue Sep 22 03:54:21 2026 +0900
feat(cpp): add functions related to consumers and offsets to high level client (#4180)
---
.../features/step_definitions/background_steps.cpp | 3 +-
foreign/cpp/.clang-tidy | 49 +
foreign/cpp/BUILD.bazel | 1 +
foreign/cpp/MODULE.bazel | 1 +
foreign/cpp/include/iggy.hpp | 668 +++++-
foreign/cpp/src/client.cpp | 97 +
foreign/cpp/src/type_conversions.cpp | 68 +-
foreign/cpp/tests/e2e/client.cpp | 2 -
foreign/cpp/tests/e2e/consumer_group.cpp | 2337 ++++++++++++++------
foreign/cpp/tests/unit/unit_tests.cpp | 35 +-
licenserc.toml | 1 +
11 files changed, 2598 insertions(+), 664 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/.clang-tidy b/foreign/cpp/.clang-tidy
new file mode 100644
index 000000000..04707277d
--- /dev/null
+++ b/foreign/cpp/.clang-tidy
@@ -0,0 +1,49 @@
+Checks: [
+ '*',
+ '-altera-unroll-loops',
+ '-bugprone-easily-swappable-parameters',
+ '-fuchsia-default-arguments-calls',
+ '-fuchsia-default-arguments-declarations',
+ '-fuchsia-trailing-return',
+ '-llvm-header-guard',
+ '-llvmlibc-callee-namespace',
+ '-llvmlibc-implementation-in-namespace',
+ '-llvmlibc-inline-function-decl',
+ '-llvmlibc-restrict-system-libc-headers',
+ '-misc-include-cleaner',
+ '-modernize-return-braced-init-list',
+ '-modernize-use-trailing-return-type',
+ '-cppcoreguidelines-rvalue-reference-param-not-moved'
+]
+
+WarningsAsErrors: '*'
+
+HeaderFileExtensions:
+- 'hpp'
+
+ImplementationFileExtensions:
+- 'cpp'
+
+HeaderFilterRegex: '.*\.hpp$'
+
+CheckOptions:
+ readability-identifier-naming.ClassCase: CamelCase
+ readability-identifier-naming.EnumCase: CamelCase
+ readability-identifier-naming.LocalVariableCase: lower_case
+ readability-identifier-naming.StaticConstantCase: aNy_CasE
+ readability-identifier-naming.MemberCase: lower_case
+ readability-identifier-naming.MemberSuffix: '_'
+ readability-identifier-naming.PrivateMemberSuffix: '_'
+ readability-identifier-naming.ProtectedMemberSuffix: '_'
+ readability-identifier-naming.PublicMemberCase: lower_case
+ readability-identifier-naming.PublicMemberSuffix: '_'
+ readability-identifier-naming.MethodCase: CamelCase
+ readability-identifier-naming.ParameterPackCase: lower_case
+ readability-identifier-naming.StructCase: CamelCase
+ readability-identifier-naming.TemplateTemplateParameterCase: CamelCase
+ readability-identifier-naming.TemplateParameterCase: lower_case
+ readability-identifier-naming.TypeTemplateParameterCase: CamelCase
+ readability-identifier-naming.TypedefCase: CamelCase
+ readability-identifier-naming.UnionCase: CamelCase
+ readability-identifier-length.MinimumParameterNameLength: 2
+ readability-identifier-length.MinimumVariableNameLength: 2
diff --git a/foreign/cpp/BUILD.bazel b/foreign/cpp/BUILD.bazel
index 286cce09e..1d3d2e8d5 100644
--- a/foreign/cpp/BUILD.bazel
+++ b/foreign/cpp/BUILD.bazel
@@ -139,6 +139,7 @@ cc_library(
visibility = ["//visibility:public"],
deps = [
":iggy_cpp_static",
+ "@abseil-cpp//absl/numeric:int128",
],
)
diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel
index 3401589b3..681b1897d 100644
--- a/foreign/cpp/MODULE.bazel
+++ b/foreign/cpp/MODULE.bazel
@@ -25,6 +25,7 @@ bazel_dep(name = "platforms", version = "1.1.0")
bazel_dep(name = "googletest", version = "1.18.0.bcr.1")
bazel_dep(name = "cucumber-cpp", version = "0.8.0.bcr.1")
bazel_dep(name = "rules_rust", version = "0.74.0")
+bazel_dep(name = "abseil-cpp", version = "20260526.0")
rust_host_tools = use_extension("@rules_rust//rust:extensions.bzl", "rust_host_tools")
rust_host_tools.host_tools(
diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp
index 230905fb5..69ad1c4f0 100644
--- a/foreign/cpp/include/iggy.hpp
+++ b/foreign/cpp/include/iggy.hpp
@@ -37,10 +37,21 @@
#include <variant>
#include <vector>
+#if defined(__GNUC__)
+# pragma GCC diagnostic push
+# pragma GCC diagnostic ignored "-Wpedantic"
+#endif
+#include "absl/numeric/int128.h"
+#if defined(__GNUC__)
+# pragma GCC diagnostic pop
+#endif
+
#include "lib.rs.h"
namespace iggy {
+class Consumer;
+class ConsumerOffsetInfo;
class IggyBlockingClient;
class LoginInfo;
class Partition;
@@ -48,6 +59,11 @@ class Topic;
class TopicDetails;
class Stream;
class StreamDetails;
+class ConsumerGroup;
+class ConsumerGroupDetails;
+class ConsumerGroupMember;
+class IggyMessagePolled;
+class IggyMessageToSend;
namespace detail {
/** @brief Internal base for string-backed option types. */
@@ -55,7 +71,11 @@ template <typename Tag>
class StringTag {
protected:
explicit StringTag(std::string value) : value_(std::move(value)) {}
- ~StringTag() = default;
+ ~StringTag() = default;
+ StringTag(const StringTag &) = default;
+ StringTag(StringTag &&) = default;
+ StringTag &operator=(const StringTag &) = default;
+ StringTag &operator=(StringTag &&) = default;
[[nodiscard]] std::string_view Value() const { return value_; }
@@ -128,7 +148,8 @@ class LoginInfo final {
*/
class Identifier final {
public:
- enum class Kind { Numeric, String };
+ static constexpr std::size_t kMaxIdentifierLength = 255;
+ enum class Kind : std::uint8_t { Numeric, String };
/**
* @brief Creates a numeric identifier.
@@ -144,7 +165,7 @@ class Identifier final {
* @throws IggyException if @p name is empty or exceeds 255 bytes.
*/
static Identifier String(std::string name) {
- if (name.empty() || name.size() > 255) {
+ if (name.empty() || name.size() > kMaxIdentifierLength) {
throw IggyException("Identifier name must contain 1 to 255 bytes");
}
return Identifier(Kind::String, std::move(name));
@@ -167,7 +188,7 @@ class Identifier final {
private:
Identifier(Kind kind, std::variant<std::uint32_t, std::string> value) : kind_(kind), value_(std::move(value)) {}
- ffi::Identifier ToFfi() const;
+ [[nodiscard]] ffi::Identifier ToFfi() const;
friend class IggyBlockingClient;
@@ -175,6 +196,99 @@ class Identifier final {
std::variant<std::uint32_t, std::string> value_;
};
+/**
+ * @brief Identifies the owner of a stored consumer offset.
+ *
+ * A consumer offset belongs either to an individual consumer or to a consumer
+ * group. Create a value with Single() or Group(), then pass it to the consumer
+ * offset operations on IggyBlockingClient.
+ */
+class Consumer final {
+ public:
+ enum class Kind : std::uint8_t { 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 the kind of consumer represented by this value.
+ * @return Kind::Single for an individual consumer or Kind::Group for a
+ * consumer group.
+ */
+ [[nodiscard]] Kind Type() const noexcept { return kind_; }
+
+ /**
+ * @brief Returns the consumer or consumer group identifier.
+ * @return Identifier owned by this value. The reference remains valid while
+ * this Consumer remains alive.
+ */
+ [[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 Snapshot of a consumer offset and its partition state.
+ *
+ * GetConsumerOffset() returns this value for an individual consumer or a
+ * consumer group. The partition's current offset can advance immediately after
+ * the request completes, while the stored offset changes only when explicitly
+ * stored or deleted.
+ */
+class ConsumerOffsetInfo final {
+ public:
+ /**
+ * @brief Returns the partition associated with the stored offset.
+ * @return Numeric partition ID.
+ */
+ [[nodiscard]] std::uint32_t PartitionId() const noexcept { return partition_id_; }
+
+ /**
+ * @brief Returns the partition's current message offset.
+ * @return Current message offset observed by the server for this request.
+ */
+ [[nodiscard]] std::uint64_t CurrentOffset() const noexcept { return current_offset_; }
+
+ /**
+ * @brief Returns the offset stored for the consumer identity.
+ * @return Stored consumer offset observed by the server for this request.
+ */
+ [[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.
*
@@ -275,12 +389,185 @@ class HeaderEntry final {
static HeaderEntry FromFfi(ffi::HeaderEntry entry);
+ friend class IggyMessagePolled;
friend class ResourceOptions;
HeaderField key_;
HeaderField value_;
};
+/**
+ * @brief Message payload and user headers prepared for sending.
+ *
+ * Create() owns the supplied payload and headers. Validation is deferred until
+ * the message is sent. A valid payload contains between 1 and 64,000,000 bytes,
+ * and the encoded user headers occupy no more than 100,000 bytes. Header keys
+ * must be unique. Header insertion order is not preserved during transmission;
+ * headers are ordered by their typed keys.
+ *
+ * The message ID is application-defined and defaults to zero. IDs do not need
+ * to be unique.
+ */
+class IggyMessageToSend final {
+ public:
+ /**
+ * @brief Creates a message for a send operation.
+ * @param payload Binary message payload.
+ * @param user_headers Optional typed user headers. Keys must be unique.
+ * @param id Application-defined message ID.
+ * @return Message owning @p payload and @p user_headers.
+ * @note Payload and header constraints are validated when the message is
+ * sent, not by this function.
+ */
+ static IggyMessageToSend Create(std::vector<std::uint8_t> payload,
+ std::vector<HeaderEntry> user_headers = {},
+ absl::uint128 id = 0) {
+ return IggyMessageToSend(id, std::move(payload), std::move(user_headers));
+ }
+
+ /**
+ * @brief Returns the application-defined message ID.
+ * @return Message ID supplied to Create(), or zero when omitted.
+ */
+ [[nodiscard]] absl::uint128 Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the binary message payload.
+ * @return Payload owned by this value. The reference remains valid while
+ * this IggyMessageToSend remains alive.
+ */
+ [[nodiscard]] const std::vector<std::uint8_t> &Payload() const noexcept { return payload_; }
+
+ /**
+ * @brief Returns the typed user headers.
+ * @return Headers owned by this value in their original insertion order.
+ * The reference remains valid while this IggyMessageToSend remains
+ * alive.
+ */
+ [[nodiscard]] const std::vector<HeaderEntry> &UserHeaders() const noexcept { return user_headers_; }
+
+ private:
+ IggyMessageToSend(absl::uint128 id, std::vector<std::uint8_t> payload, std::vector<HeaderEntry> user_headers)
+ : id_(id), payload_(std::move(payload)), user_headers_(std::move(user_headers)) {}
+
+ [[nodiscard]] ffi::IggyMessageToSend ToFfi() const;
+
+ friend class IggyBlockingClient;
+
+ absl::uint128 id_;
+ std::vector<std::uint8_t> payload_;
+ std::vector<HeaderEntry> user_headers_;
+};
+
+/**
+ * @brief Message and metadata returned by a poll operation.
+ *
+ * This value owns its payload and decoded user headers. Header entries are
+ * returned in their encoded order. Malformed encoded headers are reported as
+ * an empty collection rather than making the message unreadable.
+ */
+class IggyMessagePolled final {
+ public:
+ /**
+ * @brief Returns the stored message checksum.
+ * @return Checksum covering the message fields after the checksum field.
+ */
+ [[nodiscard]] std::uint64_t Checksum() const noexcept { return checksum_; }
+
+ /**
+ * @brief Returns the application-defined message ID.
+ * @return Message ID supplied when the message was sent.
+ */
+ [[nodiscard]] absl::uint128 Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the message offset within its partition.
+ * @return Offset assigned by the server.
+ */
+ [[nodiscard]] std::uint64_t Offset() const noexcept { return offset_; }
+
+ /**
+ * @brief Returns the timestamp assigned when the message was stored.
+ * @return Server timestamp in microseconds since the Unix epoch.
+ */
+ [[nodiscard]] std::uint64_t Timestamp() const noexcept { return timestamp_; }
+
+ /**
+ * @brief Returns the timestamp recorded when the message was created.
+ * @return Origin timestamp in microseconds since the Unix epoch.
+ */
+ [[nodiscard]] std::uint64_t OriginTimestamp() const noexcept { return origin_timestamp_; }
+
+ /**
+ * @brief Returns the encoded size of the user-header section.
+ * @return Encoded user-header length in bytes.
+ */
+ [[nodiscard]] std::uint32_t UserHeadersLength() const noexcept { return user_headers_length_; }
+
+ /**
+ * @brief Returns the payload length recorded in the message header.
+ * @return Payload length in bytes.
+ */
+ [[nodiscard]] std::uint32_t PayloadLength() const noexcept { return payload_length_; }
+
+ /**
+ * @brief Returns the message header's reserved field.
+ * @return Reserved value, currently zero.
+ */
+ [[nodiscard]] std::uint64_t Reserved() const noexcept { return reserved_; }
+
+ /**
+ * @brief Returns the binary message payload.
+ * @return Payload owned by this value. The reference remains valid while
+ * this IggyMessagePolled remains alive.
+ */
+ [[nodiscard]] const std::vector<std::uint8_t> &Payload() const noexcept { return payload_; }
+
+ /**
+ * @brief Returns the decoded typed user headers.
+ * @return Headers owned by this value in their encoded order. The reference
+ * remains valid while this IggyMessagePolled remains alive.
+ */
+ [[nodiscard]] const std::vector<HeaderEntry> &UserHeaders() const noexcept { return user_headers_; }
+
+ private:
+ IggyMessagePolled(std::uint64_t checksum,
+ absl::uint128 id,
+ std::uint64_t offset,
+ std::uint64_t timestamp,
+ std::uint64_t origin_timestamp,
+ std::uint32_t user_headers_length,
+ std::uint32_t payload_length,
+ std::uint64_t reserved,
+ std::vector<std::uint8_t> payload,
+ std::vector<HeaderEntry> user_headers)
+ : checksum_(checksum),
+ id_(id),
+ offset_(offset),
+ timestamp_(timestamp),
+ origin_timestamp_(origin_timestamp),
+ user_headers_length_(user_headers_length),
+ payload_length_(payload_length),
+ reserved_(reserved),
+ payload_(std::move(payload)),
+ user_headers_(std::move(user_headers)) {}
+
+ static IggyMessagePolled FromFfi(ffi::IggyMessagePolled message);
+
+ friend class IggyBlockingClient;
+
+ std::uint64_t checksum_;
+ absl::uint128 id_;
+ std::uint64_t offset_;
+ std::uint64_t timestamp_;
+ std::uint64_t origin_timestamp_;
+ std::uint32_t user_headers_length_;
+ std::uint32_t payload_length_;
+ std::uint64_t reserved_;
+ std::vector<std::uint8_t> payload_;
+ std::vector<HeaderEntry> user_headers_;
+};
+
/**
* @brief Options recorded for a stream or topic.
*
@@ -834,6 +1121,157 @@ class Stream final {
ResourceOptions options_;
};
+/**
+ * @brief Snapshot of a consumer group member and its partition assignments.
+ *
+ * ConsumerGroupDetails contains one of these values for every member observed
+ * by the server. Membership and partition assignments can change immediately
+ * after the request completes.
+ */
+class ConsumerGroupMember final {
+ public:
+ /**
+ * @brief Returns the numeric ID of the consumer group member.
+ * @return Numeric member ID assigned by the server.
+ */
+ [[nodiscard]] std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @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_; }
+
+ /**
+ * @brief Returns the partitions assigned to this member.
+ * @return Partition IDs owned by this value. The reference remains valid
+ * while this ConsumerGroupMember remains alive.
+ */
+ [[nodiscard]] const std::vector<std::uint32_t> &Partitions() const noexcept { return partitions_; }
+
+ private:
+ ConsumerGroupMember(std::uint32_t id, std::uint32_t partitions_count, std::vector<std::uint32_t> partitions)
+ : id_(id), partitions_count_(partitions_count), partitions_(std::move(partitions)) {}
+
+ static ConsumerGroupMember FromFfi(ffi::ConsumerGroupMember member);
+
+ friend class ConsumerGroupDetails;
+
+ std::uint32_t id_;
+ std::uint32_t partitions_count_;
+ std::vector<std::uint32_t> partitions_;
+};
+
+/**
+ * @brief Snapshot of consumer group metadata.
+ *
+ * GetConsumerGroups() returns one summary for each consumer group observed in
+ * a topic. Use GetConsumerGroup() when individual member and partition
+ * assignment details are needed.
+ */
+class ConsumerGroup final {
+ public:
+ /**
+ * @brief Returns the numeric ID assigned to the consumer group.
+ * @return Numeric consumer group ID.
+ */
+ [[nodiscard]] std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the consumer group name.
+ * @return Name owned by this value.
+ */
+ [[nodiscard]] const std::string &Name() const noexcept { return name_; }
+
+ /**
+ * @brief Returns the number of partitions consumed by the group.
+ * @return Partition count observed by the server for this request.
+ */
+ [[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return partitions_count_; }
+
+ /**
+ * @brief Returns the number of members in the group.
+ * @return Member count observed by the server for this request.
+ */
+ [[nodiscard]] std::uint32_t MembersCount() const noexcept { return members_count_; }
+
+ private:
+ ConsumerGroup(std::uint32_t id, std::string name, std::uint32_t partitions_count, std::uint32_t members_count)
+ : id_(id), name_(std::move(name)), partitions_count_(partitions_count), members_count_(members_count) {}
+
+ static ConsumerGroup FromFfi(ffi::ConsumerGroup group);
+
+ friend class IggyBlockingClient;
+
+ std::uint32_t id_;
+ std::string name_;
+ std::uint32_t partitions_count_;
+ std::uint32_t members_count_;
+};
+
+/**
+ * @brief Snapshot of consumer group metadata and member details.
+ *
+ * CreateConsumerGroup() and GetConsumerGroup() return this value. Membership
+ * and partition assignments can change immediately after the request
+ * completes.
+ */
+class ConsumerGroupDetails final {
+ public:
+ /**
+ * @brief Returns the numeric ID assigned to the consumer group.
+ * @return Numeric consumer group ID.
+ */
+ [[nodiscard]] std::uint32_t Id() const noexcept { return id_; }
+
+ /**
+ * @brief Returns the consumer group name.
+ * @return Name owned by this value.
+ */
+ [[nodiscard]] const std::string &Name() const noexcept { return name_; }
+
+ /**
+ * @brief Returns the number of partitions consumed by the group.
+ * @return Partition count observed by the server for this request.
+ */
+ [[nodiscard]] std::uint32_t PartitionsCount() const noexcept { return partitions_count_; }
+
+ /**
+ * @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_; }
+
+ /**
+ * @brief Returns the consumer group members and their partition assignments.
+ * @return Member details owned by this value. The reference remains valid
+ * while this ConsumerGroupDetails remains alive.
+ */
+ [[nodiscard]] const std::vector<ConsumerGroupMember> &Members() const noexcept { return members_; }
+
+ private:
+ ConsumerGroupDetails(std::uint32_t id,
+ std::string name,
+ std::uint32_t partitions_count,
+ std::uint32_t members_count,
+ std::vector<ConsumerGroupMember> members)
+ : id_(id),
+ name_(std::move(name)),
+ partitions_count_(partitions_count),
+ members_count_(members_count),
+ members_(std::move(members)) {}
+
+ static ConsumerGroupDetails FromFfi(ffi::ConsumerGroupDetails group);
+
+ friend class IggyBlockingClient;
+
+ std::uint32_t id_;
+ std::string name_;
+ std::uint32_t partitions_count_;
+ std::uint32_t members_count_;
+ std::vector<ConsumerGroupMember> members_;
+};
+
/**
* @brief Compression algorithm used for topic messages.
*
@@ -1033,7 +1471,7 @@ class Expiry final {
std::uint64_t expiry_value_;
};
-enum class Durability { Replicated, Persisted };
+enum class Durability : std::uint8_t { Replicated, Persisted };
constexpr std::string_view to_string(const Durability durability) {
switch (durability) {
@@ -1985,6 +2423,204 @@ class IggyBlockingClient final {
*/
void DeletePartitions(const Identifier &stream, const Identifier &topic, std::uint32_t partitions_count);
+ /**
+ * @brief Creates a consumer group for a topic.
+ *
+ * The group name must be unique within the topic, non-empty, and no more
+ * than 255 UTF-8 bytes. The new group initially has no members.
+ *
+ * The VSR server assigns consumer group IDs monotonically. Deleting a
+ * group and recreating it with the same name is allowed, but the recreated
+ * group receives a new ID rather than reusing the deleted group's ID.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Parent topic, addressed by numeric ID or name.
+ * @param name Unique consumer group name within @p topic.
+ * @return Details of the newly created consumer group.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier or the name is invalid; the stream or topic does
+ * not exist; the name is already in use; the caller lacks
+ * stream- or topic-management permission; or the request fails.
+ */
+ ConsumerGroupDetails CreateConsumerGroup(const Identifier &stream, const Identifier &topic, std::string name);
+
+ /**
+ * @brief Retrieves one consumer group and its current members.
+ *
+ * The returned details are a snapshot. Membership and partition
+ * assignments can change immediately after this call returns.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Parent topic, addressed by numeric ID or name.
+ * @param group Consumer group to retrieve, addressed by numeric ID or name.
+ * @return Consumer group metadata and member details.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier is invalid; the stream, topic, or consumer group
+ * does not exist; the caller lacks read permission; or the
+ * metadata read fails.
+ */
+ ConsumerGroupDetails GetConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group);
+
+ /**
+ * @brief Lists consumer group summaries for a topic.
+ *
+ * The summaries include member and partition counts but omit individual
+ * member details. Use GetConsumerGroup() to retrieve those details.
+ *
+ * The VSR server reports a missing parent stream or topic as an error. This
+ * differs from the legacy server, which returned an empty list, so an empty
+ * result does not establish whether the parent resources exist across
+ * server implementations.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Parent topic, addressed by numeric ID or name.
+ * @return Consumer group summaries for the requested topic.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier is invalid; the stream or topic does not exist;
+ * the caller lacks read permission; or the metadata read fails.
+ */
+ std::vector<ConsumerGroup> GetConsumerGroups(const Identifier &stream, const Identifier &topic);
+
+ /**
+ * @brief Deletes a consumer group from a topic.
+ *
+ * A failed or unknown transport outcome can leave the deletion committed.
+ * Query the topic's consumer groups before retrying this request.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Parent topic, addressed by numeric ID or name.
+ * @param group Consumer group to delete, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier is invalid; the stream, topic, or consumer group
+ * does not exist; the caller lacks stream- or topic-management
+ * permission; or the request fails.
+ */
+ void DeleteConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group);
+
+ /**
+ * @brief Joins the current client to a consumer group.
+ *
+ * The server assigns topic partitions among the group's members. Joining
+ * the same group again does not add a second membership for this client.
+ * Joining consumer groups over HTTP is not supported.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Parent topic, addressed by numeric ID or name.
+ * @param group Consumer group to join, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier is invalid; the stream, topic, or consumer group
+ * does not exist; the caller lacks read permission; the transport
+ * does not support group membership; or the request fails.
+ */
+ void JoinConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group);
+
+ /**
+ * @brief Removes the current client from a consumer group.
+ *
+ * The server reassigns partitions among the remaining group members.
+ * The client must currently belong to the group; leaving twice or leaving
+ * without first joining fails. Leaving consumer groups over HTTP is not
+ * supported.
+ *
+ * @param stream Parent stream, addressed by numeric ID or name.
+ * @param topic Parent topic, addressed by numeric ID or name.
+ * @param group Consumer group to leave, addressed by numeric ID or name.
+ * @throws IggyException if the client is unavailable or unauthenticated;
+ * an identifier is invalid; the stream, topic, or consumer group
+ * does not exist; this client is not a member; the caller lacks
+ * read permission; the transport does not support group
+ * membership; or the request fails.
+ */
+ void LeaveConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group);
+
+ /**
+ * @brief Stores an offset for a consumer or consumer group.
+ *
+ * The server accepts offsets from zero through the partition's current
+ * offset, inclusive. It rejects every offset for an empty partition and
+ * any offset beyond the current offset. Storing another value for the same
+ * consumer and partition replaces the previous value.
+ *
+ * For a consumer group, the group must exist and the current client must
+ * own @p partition_id in that group. This ownership fence does not apply to
+ * individual consumers.
+ *
+ * @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 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.
+ * @param offset Message offset to store.
+ * @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::optional<std::uint32_t> partition_id,
+ std::uint64_t offset);
+
+ /**
+ * @brief Retrieves the stored offset for a consumer or consumer group.
+ *
+ * This method throws IggyException when no offset has been stored. A
+ * consumer group offset can be read by an authenticated caller with poll
+ * permission even when that client is not a member of the group.
+ *
+ * An offset created by an auto-commit poll is visible through this method.
+ * A local auto-commit cursor can be visible before its durable store has
+ * committed.
+ *
+ * @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, 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
+ * unauthenticated; the caller lacks permission; or the request
+ * fails.
+ */
+ ConsumerOffsetInfo GetConsumerOffset(const Consumer &consumer,
+ const Identifier &stream,
+ const Identifier &topic,
+ std::optional<std::uint32_t> partition_id = std::nullopt);
+
+ /**
+ * @brief Deletes the stored offset for a consumer or consumer group.
+ *
+ * Deletion is not idempotent: deleting an offset that was never stored, or
+ * deleting the same offset again, fails. A failed or unknown transport
+ * outcome can leave the deletion committed, in which case a retry can fail
+ * because the offset is already absent.
+ *
+ * For a consumer group, the group must exist and the current client must
+ * own @p partition_id in that group. This ownership fence does not apply to
+ * individual consumers.
+ *
+ * @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, 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
+ * fails.
+ */
+ void DeleteConsumerOffset(const Consumer &consumer,
+ const Identifier &stream,
+ const Identifier &topic,
+ std::optional<std::uint32_t> partition_id = std::nullopt);
+
private:
explicit IggyBlockingClient(ffi::Client *client);
@@ -2183,21 +2819,21 @@ class IggyBlockingClient::Builder final {
* @return Configured client.
* @throws IggyException if validation or client creation fails.
*/
- IggyBlockingClient Build() const;
+ [[nodiscard]] IggyBlockingClient Build() const;
private:
- std::string server_address_{};
+ std::string server_address_;
ffi::AutoLoginKind auto_login_kind_{ffi::AutoLoginKind::Disabled};
- std::string auto_login_username_{};
- std::string auto_login_password_{};
- std::string personal_access_token_{};
- std::optional<std::uint32_t> reconnection_max_retries_{};
- std::optional<std::uint64_t> reconnection_interval_micros_{};
- std::optional<std::uint64_t> reestablish_after_micros_{};
+ std::string auto_login_username_;
+ std::string auto_login_password_;
+ std::string personal_access_token_;
+ std::optional<std::uint32_t> reconnection_max_retries_;
+ std::optional<std::uint64_t> reconnection_interval_micros_;
+ std::optional<std::uint64_t> reestablish_after_micros_;
bool tls_enabled_{};
- std::string tls_domain_{};
- std::string tls_ca_file_{};
- std::optional<bool> tls_validate_certificate_{};
+ std::string tls_domain_;
+ std::string tls_ca_file_;
+ std::optional<bool> tls_validate_certificate_;
bool no_delay_{};
};
diff --git a/foreign/cpp/src/client.cpp b/foreign/cpp/src/client.cpp
index 3e56a4baf..d5a3ca3f4 100644
--- a/foreign/cpp/src/client.cpp
+++ b/foreign/cpp/src/client.cpp
@@ -252,6 +252,103 @@ void IggyBlockingClient::DeletePartitions(const Identifier &stream,
});
}
+ConsumerGroupDetails IggyBlockingClient::CreateConsumerGroup(const Identifier &stream,
+ const Identifier &topic,
+ std::string name) {
+ return RethrowAsIggyException([this, &stream, &topic, &name] {
+ return ConsumerGroupDetails::FromFfi(Handle()->create_consumer_group(stream.ToFfi(), topic.ToFfi(), name));
+ });
+}
+
+ConsumerGroupDetails IggyBlockingClient::GetConsumerGroup(const Identifier &stream,
+ const Identifier &topic,
+ const Identifier &group) {
+ return RethrowAsIggyException([this, &stream, &topic, &group] {
+ return ConsumerGroupDetails::FromFfi(
+ Handle()->get_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi()));
+ });
+}
+
+std::vector<ConsumerGroup> IggyBlockingClient::GetConsumerGroups(const Identifier &stream, const Identifier &topic) {
+ return RethrowAsIggyException([this, &stream, &topic] {
+ std::vector<ConsumerGroup> groups;
+ auto ffi_groups = Handle()->get_consumer_groups(stream.ToFfi(), topic.ToFfi());
+ groups.reserve(ffi_groups.size());
+ for (auto &group : ffi_groups) {
+ groups.push_back(ConsumerGroup::FromFfi(std::move(group)));
+ }
+ return groups;
+ });
+}
+
+void IggyBlockingClient::DeleteConsumerGroup(const Identifier &stream,
+ const Identifier &topic,
+ const Identifier &group) {
+ RethrowAsIggyException([this, &stream, &topic, &group] {
+ Handle()->delete_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi());
+ });
+}
+
+void IggyBlockingClient::JoinConsumerGroup(const Identifier &stream, const Identifier &topic, const Identifier &group) {
+ RethrowAsIggyException([this, &stream, &topic, &group] {
+ Handle()->join_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi());
+ });
+}
+
+void IggyBlockingClient::LeaveConsumerGroup(const Identifier &stream,
+ const Identifier &topic,
+ const Identifier &group) {
+ RethrowAsIggyException([this, &stream, &topic, &group] {
+ Handle()->leave_consumer_group(stream.ToFfi(), topic.ToFfi(), group.ToFfi());
+ });
+}
+
+void IggyBlockingClient::StoreConsumerOffset(const Consumer &consumer,
+ const Identifier &stream,
+ const Identifier &topic,
+ const std::optional<std::uint32_t> partition_id,
+ const std::uint64_t offset) {
+ RethrowAsIggyException([this, &consumer, &stream, &topic, offset, 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);
+ 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::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(), ffi_partition_id, std::string(consumer.KindName()), consumer.Id().ToFfi()));
+ });
+}
+
+void IggyBlockingClient::DeleteConsumerOffset(const Consumer &consumer,
+ const Identifier &stream,
+ const Identifier &topic,
+ const std::optional<std::uint32_t> partition_id) {
+ 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);
+ Handle()->delete_consumer_offset(stream.ToFfi(), topic.ToFfi(), ffi_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 31a1d33b9..166752279 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()));
@@ -51,6 +55,47 @@ HeaderEntry HeaderEntry::FromFfi(ffi::HeaderEntry entry) {
return HeaderEntry(HeaderField::FromFfi(std::move(entry.key)), HeaderField::FromFfi(std::move(entry.value)));
}
+ffi::IggyMessageToSend IggyMessageToSend::ToFfi() const {
+ ffi::IggyMessageToSend ffi_message;
+ ffi_message.id_lo = absl::Uint128Low64(id_);
+ ffi_message.id_hi = absl::Uint128High64(id_);
+ ffi_message.payload.reserve(payload_.size());
+ for (const auto byte : payload_) {
+ ffi_message.payload.push_back(byte);
+ }
+ ffi_message.user_headers.reserve(user_headers_.size());
+ for (const auto &entry : user_headers_) {
+ const auto &key = entry.Key();
+ const auto &value = entry.Value();
+ ffi::HeaderEntry ffi_entry;
+ ffi_entry.key.kind = static_cast<std::uint8_t>(key.Kind());
+ ffi_entry.key.value.reserve(key.Value().size());
+ for (const auto byte : key.Value()) {
+ ffi_entry.key.value.push_back(byte);
+ }
+ ffi_entry.value.kind = static_cast<std::uint8_t>(value.Kind());
+ ffi_entry.value.value.reserve(value.Value().size());
+ for (const auto byte : value.Value()) {
+ ffi_entry.value.value.push_back(byte);
+ }
+ ffi_message.user_headers.push_back(std::move(ffi_entry));
+ }
+ return ffi_message;
+}
+
+IggyMessagePolled IggyMessagePolled::FromFfi(ffi::IggyMessagePolled message) {
+ std::vector<HeaderEntry> user_headers;
+ user_headers.reserve(message.user_headers.size());
+ for (auto &entry : message.user_headers) {
+ user_headers.push_back(HeaderEntry::FromFfi(std::move(entry)));
+ }
+
+ return IggyMessagePolled(
+ message.checksum, absl::MakeUint128(message.id_hi, message.id_lo), message.offset, message.timestamp,
+ message.origin_timestamp, message.user_headers_length, message.payload_length, message.reserved,
+ std::vector<std::uint8_t>(message.payload.begin(), message.payload.end()), std::move(user_headers));
+}
+
ResourceOptions ResourceOptions::FromFfi(rust::Vec<ffi::HeaderEntry> explicit_entries,
rust::Vec<ffi::HeaderEntry> derived_entries) {
std::map<std::string, HeaderField> explicit_options;
@@ -87,7 +132,7 @@ TopicDetails TopicDetails::FromFfi(ffi::TopicDetails topic) {
std::vector<Partition> partitions;
partitions.reserve(topic.partitions.size());
for (auto &partition : topic.partitions) {
- partitions.push_back(Partition::FromFfi(std::move(partition)));
+ partitions.push_back(Partition::FromFfi(partition));
}
return TopicDetails(topic.id, topic.created_at, std::string(topic.name.c_str(), topic.name.size()),
@@ -115,4 +160,25 @@ Stream Stream::FromFfi(ffi::Stream stream) {
ResourceOptions::FromFfi(std::move(stream.options), rust::Vec<ffi::HeaderEntry>{}));
}
+ConsumerGroupMember ConsumerGroupMember::FromFfi(ffi::ConsumerGroupMember member) {
+ return ConsumerGroupMember(member.id, member.partitions_count,
+ std::vector<std::uint32_t>(member.partitions.begin(), member.partitions.end()));
+}
+
+ConsumerGroup ConsumerGroup::FromFfi(ffi::ConsumerGroup group) {
+ return ConsumerGroup(group.id, std::string(group.name.c_str(), group.name.size()), group.partitions_count,
+ group.members_count);
+}
+
+ConsumerGroupDetails ConsumerGroupDetails::FromFfi(ffi::ConsumerGroupDetails group) {
+ std::vector<ConsumerGroupMember> members;
+ members.reserve(group.members.size());
+ for (auto &member : group.members) {
+ members.push_back(ConsumerGroupMember::FromFfi(std::move(member)));
+ }
+
+ return ConsumerGroupDetails(group.id, std::string(group.name.c_str(), group.name.size()), group.partitions_count,
+ group.members_count, std::move(members));
+}
+
} // namespace iggy
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 7099e5f22..404932cf5 100644
--- a/foreign/cpp/tests/e2e/consumer_group.cpp
+++ b/foreign/cpp/tests/e2e/consumer_group.cpp
@@ -22,280 +22,280 @@
#include <gtest/gtest.h>
+#include "iggy.hpp"
#include "lib.rs.h"
#include "tests/e2e/test_helpers.hpp"
-class LowLevelE2E_ConsumerGroup : public E2ETestFixture {};
+class E2E_ConsumerGroup : public E2ETestFixture {};
-TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupSucceeds) {
+TEST_F(E2E_ConsumerGroup, CreateConsumerGroupSucceeds) {
RecordProperty("description", "Creates a consumer group successfully for an existing stream and topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
ASSERT_NO_THROW({
- const auto group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name);
+ const auto group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name);
TrackConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_EQ(group.name, group_name);
- ASSERT_EQ(group.members_count, 0u);
- ASSERT_TRUE(group.members.empty());
+ ASSERT_EQ(group.Name(), group_name);
+ ASSERT_EQ(group.MembersCount(), 0u);
+ ASSERT_TRUE(group.Members().empty());
});
}
-TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupOnNonExistentResourcesThrows) {
+TEST_F(E2E_ConsumerGroup, CreateConsumerGroupOnNonExistentResourcesThrows) {
RecordProperty("description", "Rejects creating a consumer group on streams or topics that do not exist.");
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();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
-
- ASSERT_THROW(client->create_consumer_group(make_string_identifier(missing_stream_name),
- make_string_identifier(topic_name), GetRandomName()),
- std::exception);
- ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(missing_topic_name), GetRandomName()),
- std::exception);
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+
+ ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(missing_stream_name),
+ iggy::Identifier::String(topic_name), GetRandomName()),
+ iggy::IggyException);
+ ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(missing_topic_name), GetRandomName()),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupTwiceOnSameInputThrows) {
+TEST_F(E2E_ConsumerGroup, CreateConsumerGroupTwiceOnSameInputThrows) {
RecordProperty("description", "Rejects creating the same consumer group twice for the same stream and topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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_THROW(client->create_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- group_name),
- std::exception);
+ ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ group_name),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupWithInvalidNamesThrows) {
+TEST_F(E2E_ConsumerGroup, CreateConsumerGroupWithInvalidNamesThrows) {
RecordProperty("description", "Rejects empty and overlong consumer group names.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
const std::string invalid_names[] = {"", std::string(256, 'a')};
for (const std::string &invalid_name : invalid_names) {
SCOPED_TRACE(invalid_name.size());
- ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), invalid_name),
- std::exception);
+ ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), invalid_name),
+ iggy::IggyException);
}
}
-TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupAfterStreamDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, CreateConsumerGroupAfterStreamDeletionThrows) {
RecordProperty("description", "Rejects creating a consumer group after deleting the stream that owned the topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->create_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- group_name),
- std::exception);
+ ASSERT_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ group_name),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, CreateConsumerGroupBeforeLoginThrows) {
+TEST_F(E2E_ConsumerGroup, CreateConsumerGroupBeforeLoginThrows) {
RecordProperty("description",
"Rejects creating a consumer group before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *setup_client = GetLoggedInClient();
- ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ auto setup_client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(setup_client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- setup_client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
-
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
-
- ASSERT_THROW(unauthenticated_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name),
- std::exception);
+ ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(unauthenticated_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupReturnsSameInfoAsCreateConsumerGroup) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupReturnsSameInfoAsCreateConsumerGroup) {
RecordProperty("description",
"Returns the same consumer group details from get_consumer_group as create_consumer_group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- const auto created_group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name);
+ const auto created_group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name);
TrackConsumerGroup(stream_name, topic_name, group_name);
- const auto fetched_group = client->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
+ const auto fetched_group =
+ client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
- ASSERT_EQ(fetched_group.id, created_group.id);
- ASSERT_EQ(fetched_group.name, created_group.name);
- ASSERT_EQ(fetched_group.partitions_count, created_group.partitions_count);
- ASSERT_EQ(fetched_group.members_count, created_group.members_count);
- ASSERT_EQ(fetched_group.members.size(), created_group.members.size());
+ ASSERT_EQ(fetched_group.Id(), created_group.Id());
+ ASSERT_EQ(fetched_group.Name(), created_group.Name());
+ ASSERT_EQ(fetched_group.PartitionsCount(), created_group.PartitionsCount());
+ ASSERT_EQ(fetched_group.MembersCount(), created_group.MembersCount());
+ ASSERT_EQ(fetched_group.Members().size(), created_group.Members().size());
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReturnsCreatedGroups) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsReturnsCreatedGroups) {
RecordProperty("description", "Returns created consumer groups for an existing stream and topic.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string first_group_name = GetRandomName();
const std::string second_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), first_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), first_group_name));
TrackConsumerGroup(stream_name, topic_name, first_group_name);
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), second_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), second_group_name));
TrackConsumerGroup(stream_name, topic_name, second_group_name);
const auto groups =
- client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
EXPECT_EQ(groups.size(), std::size_t{2});
- EXPECT_EQ(groups[0].name, first_group_name);
- EXPECT_EQ(groups[1].name, second_group_name);
+ EXPECT_EQ(groups[0].Name(), first_group_name);
+ EXPECT_EQ(groups[1].Name(), second_group_name);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsBeforeLoginThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsBeforeLoginThrows) {
RecordProperty("description", "Rejects get_consumer_groups before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *setup_client = GetLoggedInClient();
- ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ auto setup_client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(setup_client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- setup_client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
-
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
-
- ASSERT_THROW(unauthenticated_client->get_consumer_groups(make_string_identifier(stream_name),
- make_string_identifier(topic_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->get_consumer_groups(make_string_identifier(stream_name),
- make_string_identifier(topic_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->get_consumer_groups(make_string_identifier(stream_name),
- make_string_identifier(topic_name)),
- std::exception);
+ ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(unauthenticated_client.GetConsumerGroups(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.GetConsumerGroups(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.GetConsumerGroups(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupSucceeds) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupSucceeds) {
RecordProperty("description", "Joins an existing consumer group successfully.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupBeforeLoginThrows) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupBeforeLoginThrows) {
RecordProperty("description", "Rejects join_consumer_group before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *setup_client = GetLoggedInClient();
- ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ auto setup_client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(setup_client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- setup_client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
TrackConsumerGroup(stream_name, topic_name, group_name);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
-
- ASSERT_THROW(unauthenticated_client->join_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->join_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->join_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupOnNonExistentResourcesThrows) {
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(unauthenticated_client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+}
+
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupOnNonExistentResourcesThrows) {
RecordProperty("description", "Rejects join_consumer_group for streams, topics, or groups that do not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -303,261 +303,273 @@ TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupOnNonExistentResourcesThrows)
const std::string missing_stream_name = GetRandomName();
const std::string missing_topic_name = GetRandomName();
const std::string missing_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), created_group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), created_group_name));
TrackConsumerGroup(stream_name, topic_name, created_group_name);
ASSERT_THROW(
- client->join_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
+ client.JoinConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
ASSERT_THROW(
- client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
- ASSERT_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(missing_group_name)),
- std::exception);
+ client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
+ ASSERT_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(missing_group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupAfterStreamDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupAfterStreamDeletionThrows) {
RecordProperty("description", "Rejects join_consumer_group after deleting the stream that owned the group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupAfterTopicDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupAfterTopicDeletionThrows) {
RecordProperty("description", "Rejects join_consumer_group after deleting the topic that owned the group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
+ ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupReflectsInGetConsumerGroup) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupReflectsInGetConsumerGroup) {
RecordProperty("description", "Reflects a joined consumer group in get_consumer_group member details.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- iggy::ffi::ConsumerGroupDetails created_group;
- ASSERT_NO_THROW({
- created_group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name);
- });
+ const auto created_group = 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
-
- const auto fetched_group = client->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
-
- EXPECT_EQ(fetched_group.id, created_group.id);
- EXPECT_EQ(fetched_group.name, created_group.name);
- EXPECT_EQ(fetched_group.partitions_count, created_group.partitions_count);
- EXPECT_EQ(fetched_group.members_count, 1u);
- ASSERT_EQ(fetched_group.members.size(), std::size_t{1});
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ const auto fetched_group =
+ client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+
+ EXPECT_EQ(fetched_group.Id(), created_group.Id());
+ EXPECT_EQ(fetched_group.Name(), created_group.Name());
+ EXPECT_EQ(fetched_group.PartitionsCount(), created_group.PartitionsCount());
+ EXPECT_EQ(fetched_group.MembersCount(), 1u);
+ ASSERT_EQ(fetched_group.Members().size(), std::size_t{1});
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupTwiceKeepsSingleMember) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupTwiceKeepsSingleMember) {
RecordProperty("description",
"Allows joining the same consumer group twice in a row without duplicating membership.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
-
- const auto fetched_group = client->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
- EXPECT_EQ(fetched_group.members_count, 1u);
- ASSERT_EQ(fetched_group.members.size(), std::size_t{1});
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ const auto fetched_group =
+ client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+ EXPECT_EQ(fetched_group.MembersCount(), 1u);
+ ASSERT_EQ(fetched_group.Members().size(), std::size_t{1});
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupFromTwoClientsIncreasesMembersCount) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupFromTwoClientsIncreasesMembersCount) {
RecordProperty("description", "Reflects two joined clients as two members in the same consumer group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *first = GetLoggedInClient();
- iggy::ffi::Client *second = GetLoggedInClient();
+ auto first = GetLoggedInHighLevelClient();
+ auto second = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(first->create_stream(stream_name));
+ ASSERT_NO_THROW(first.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(first->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(first->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(first.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(first.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
TrackConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_NO_THROW(first->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- ASSERT_NO_THROW(second->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
-
- const auto fetched_group = first->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
- EXPECT_EQ(fetched_group.members_count, 2u);
- ASSERT_EQ(fetched_group.members.size(), std::size_t{2});
+ ASSERT_NO_THROW(first.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ ASSERT_NO_THROW(second.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ const auto fetched_group =
+ first.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+ EXPECT_EQ(fetched_group.MembersCount(), 2u);
+ ASSERT_EQ(fetched_group.Members().size(), std::size_t{2});
}
-TEST_F(LowLevelE2E_ConsumerGroup, JoinConsumerGroupThenLeaveRestoresMembersCount) {
+TEST_F(E2E_ConsumerGroup, JoinConsumerGroupThenLeaveRestoresMembersCount) {
RecordProperty("description", "Restores the consumer group member count after a client joins and then leaves.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- auto joined_group = client->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
- EXPECT_EQ(joined_group.members_count, 1u);
- ASSERT_EQ(joined_group.members.size(), std::size_t{1});
-
- ASSERT_NO_THROW(client->leave_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
-
- const auto left_group = client->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
- EXPECT_EQ(left_group.members_count, 0u);
- EXPECT_TRUE(left_group.members.empty());
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ const auto joined_group =
+ client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+ EXPECT_EQ(joined_group.MembersCount(), 1u);
+ ASSERT_EQ(joined_group.Members().size(), std::size_t{1});
+
+ ASSERT_NO_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ const auto left_group =
+ client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+ EXPECT_EQ(left_group.MembersCount(), 0u);
+ EXPECT_TRUE(left_group.Members().empty());
}
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupReducesMembersCount) {
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupReducesMembersCount) {
RecordProperty("description", "Reduces the consumer group member count after one of two joined clients leaves.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *first = GetLoggedInClient();
- iggy::ffi::Client *second = GetLoggedInClient();
+ auto first = GetLoggedInHighLevelClient();
+ auto second = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(first->create_stream(stream_name));
+ ASSERT_NO_THROW(first.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(first->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(first->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(first.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(first.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
TrackConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_NO_THROW(first->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- ASSERT_NO_THROW(second->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
-
- const auto joined_group = first->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
- EXPECT_EQ(joined_group.members_count, 2u);
- ASSERT_EQ(joined_group.members.size(), std::size_t{2});
-
- ASSERT_NO_THROW(second->leave_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
-
- const auto left_group = first->get_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name));
- EXPECT_EQ(left_group.members_count, 1u);
- ASSERT_EQ(left_group.members.size(), std::size_t{1});
+ ASSERT_NO_THROW(first.JoinConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ ASSERT_NO_THROW(second.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ const auto joined_group =
+ first.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+ EXPECT_EQ(joined_group.MembersCount(), 2u);
+ ASSERT_EQ(joined_group.Members().size(), std::size_t{2});
+
+ ASSERT_NO_THROW(second.LeaveConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ const auto left_group =
+ first.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name));
+ EXPECT_EQ(left_group.MembersCount(), 1u);
+ ASSERT_EQ(left_group.Members().size(), std::size_t{1});
}
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupBeforeLoginThrows) {
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupBeforeLoginThrows) {
RecordProperty("description", "Rejects leave_consumer_group before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *setup_client = GetLoggedInClient();
- ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ auto setup_client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(setup_client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- setup_client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
TrackConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_NO_THROW(setup_client->join_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
-
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
-
- ASSERT_THROW(unauthenticated_client->leave_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->leave_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->leave_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupOnNonExistentResourcesThrows) {
+ ASSERT_NO_THROW(setup_client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(unauthenticated_client.LeaveConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.LeaveConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.LeaveConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+}
+
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupOnNonExistentResourcesThrows) {
RecordProperty("description", "Rejects leave_consumer_group for streams, topics, or groups that do not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -565,372 +577,364 @@ TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupOnNonExistentResourcesThrows
const std::string missing_stream_name = GetRandomName();
const std::string missing_topic_name = GetRandomName();
const std::string missing_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), created_group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), created_group_name));
TrackConsumerGroup(stream_name, topic_name, created_group_name);
- ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(created_group_name)));
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(created_group_name)));
ASSERT_THROW(
- client->leave_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
+ client.LeaveConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
ASSERT_THROW(
- client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
- ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(missing_group_name)),
- std::exception);
+ client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
+ ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(missing_group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupAfterStreamDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupAfterStreamDeletionThrows) {
RecordProperty("description", "Rejects leave_consumer_group after deleting the stream that owned the group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupAfterTopicDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupAfterTopicDeletionThrows) {
RecordProperty("description", "Rejects leave_consumer_group after deleting the topic that owned the group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupTwiceThrows) {
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupTwiceThrows) {
RecordProperty("description", "Rejects leaving the same consumer group twice.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)));
- ASSERT_NO_THROW(client->leave_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
-
- ASSERT_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+ ASSERT_NO_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
+
+ ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, LeaveConsumerGroupWithoutJoiningThrows) {
+TEST_F(E2E_ConsumerGroup, LeaveConsumerGroupWithoutJoiningThrows) {
RecordProperty("description", "Rejects leaving a consumer group when the client is not a member.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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_THROW(client->leave_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.LeaveConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReflectsJoinedGroupMembersCount) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsReflectsJoinedGroupMembersCount) {
RecordProperty("description", "Reflects a joined consumer group in get_consumer_groups members_count.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string joined_group_name = GetRandomName();
const std::string other_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- iggy::ffi::ConsumerGroupDetails joined_group;
- iggy::ffi::ConsumerGroupDetails other_group;
- ASSERT_NO_THROW({
- joined_group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), joined_group_name);
- });
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), joined_group_name));
TrackConsumerGroup(stream_name, topic_name, joined_group_name);
- ASSERT_NO_THROW({
- other_group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), other_group_name);
- });
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), other_group_name));
TrackConsumerGroup(stream_name, topic_name, other_group_name);
- ASSERT_NO_THROW(client->join_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(joined_group_name)));
+ ASSERT_NO_THROW(client.JoinConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(joined_group_name)));
const auto groups =
- client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
ASSERT_EQ(groups.size(), std::size_t{2});
- EXPECT_EQ(groups[0].id, joined_group.id);
- EXPECT_EQ(groups[0].name, joined_group.name);
- EXPECT_EQ(groups[0].members_count, joined_group.members_count + 1u);
+ EXPECT_EQ(groups[0].Name(), joined_group_name);
+ EXPECT_EQ(groups[0].MembersCount(), 1u);
- EXPECT_EQ(groups[1].id, other_group.id);
- EXPECT_EQ(groups[1].name, other_group.name);
- EXPECT_EQ(groups[1].members_count, other_group.members_count);
- EXPECT_NE(groups[0].members_count, groups[1].members_count);
+ EXPECT_EQ(groups[1].Name(), other_group_name);
+ EXPECT_EQ(groups[1].MembersCount(), 0u);
+ EXPECT_NE(groups[0].MembersCount(), groups[1].MembersCount());
}
// The VSR server rejects consumer-group reads whose parent stream or topic is
// absent with the legacy typed not-found; the legacy server answered them with
// an empty list.
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentStreamThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsOnNonExistentStreamThrows) {
RecordProperty("description", "Throws when the stream does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)),
- std::exception);
+ ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsOnNonExistentTopicThrows) {
RecordProperty("description", "Throws when the topic does not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)),
- std::exception);
+ ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsIsStableAcrossBackToBackCalls) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsIsStableAcrossBackToBackCalls) {
RecordProperty("description", "Returns the same consumer groups across back-to-back get_consumer_groups calls.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string first_group_name = GetRandomName();
const std::string second_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), first_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), first_group_name));
TrackConsumerGroup(stream_name, topic_name, first_group_name);
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), second_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), second_group_name));
TrackConsumerGroup(stream_name, topic_name, second_group_name);
const auto first_groups =
- client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
const auto second_groups =
- client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
EXPECT_EQ(second_groups.size(), first_groups.size());
EXPECT_EQ(second_groups.size(), std::size_t{2});
for (std::size_t i = 0; i < first_groups.size(); ++i) {
- EXPECT_EQ(second_groups[i].id, first_groups[i].id);
- EXPECT_EQ(second_groups[i].name, first_groups[i].name);
- EXPECT_EQ(second_groups[i].partitions_count, first_groups[i].partitions_count);
- EXPECT_EQ(second_groups[i].members_count, first_groups[i].members_count);
+ EXPECT_EQ(second_groups[i].Id(), first_groups[i].Id());
+ EXPECT_EQ(second_groups[i].Name(), first_groups[i].Name());
+ EXPECT_EQ(second_groups[i].PartitionsCount(), first_groups[i].PartitionsCount());
+ EXPECT_EQ(second_groups[i].MembersCount(), first_groups[i].MembersCount());
}
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsReturnsCorrectNumberOfGroups) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsReturnsCorrectNumberOfGroups) {
RecordProperty("description", "Returns the last remaining consumer group after deleting two groups.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string deleted_group_name = GetRandomName();
const std::string other_deleted_name = GetRandomName();
const std::string remaining_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), deleted_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), deleted_group_name));
TrackConsumerGroup(stream_name, topic_name, deleted_group_name);
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), other_deleted_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), other_deleted_name));
TrackConsumerGroup(stream_name, topic_name, other_deleted_name);
- iggy::ffi::ConsumerGroupDetails remaining_group;
- ASSERT_NO_THROW({
- remaining_group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), remaining_group_name);
- });
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), remaining_group_name));
TrackConsumerGroup(stream_name, topic_name, remaining_group_name);
- ASSERT_NO_THROW(client->delete_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(deleted_group_name)));
+ ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(deleted_group_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, deleted_group_name);
- ASSERT_NO_THROW(client->delete_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(other_deleted_name)));
+ ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(other_deleted_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, other_deleted_name);
const auto groups =
- client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
ASSERT_EQ(groups.size(), std::size_t{1});
- EXPECT_EQ(groups[0].id, remaining_group.id);
- EXPECT_EQ(groups[0].name, remaining_group.name);
- EXPECT_EQ(groups[0].partitions_count, remaining_group.partitions_count);
- EXPECT_EQ(groups[0].members_count, remaining_group.members_count);
-
- ASSERT_NO_THROW(client->delete_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(remaining_group_name)));
+ EXPECT_EQ(groups[0].Name(), remaining_group_name);
+ EXPECT_EQ(groups[0].MembersCount(), 0u);
+
+ ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(remaining_group_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, remaining_group_name);
const auto groups_after_delete =
- client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name));
+ client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name));
EXPECT_TRUE(groups_after_delete.empty());
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsAfterStreamDeletionThrows) {
RecordProperty("description", "Throws after deleting the stream that owned the groups.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string first_group_name = GetRandomName();
const std::string second_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), first_group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), first_group_name));
TrackConsumerGroup(stream_name, topic_name, first_group_name);
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), second_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), second_group_name));
TrackConsumerGroup(stream_name, topic_name, second_group_name);
- ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, first_group_name);
ForgetTrackedConsumerGroup(stream_name, topic_name, second_group_name);
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)),
- std::exception);
+ ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupsAfterTopicDeletionThrows) {
RecordProperty("description", "Throws after deleting the topic that owned the groups.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string first_group_name = GetRandomName();
const std::string second_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), first_group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), first_group_name));
TrackConsumerGroup(stream_name, topic_name, first_group_name);
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), second_group_name));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), second_group_name));
TrackConsumerGroup(stream_name, topic_name, second_group_name);
- ASSERT_NO_THROW(client->delete_topic(make_string_identifier(stream_name), make_string_identifier(topic_name)));
+ ASSERT_NO_THROW(client.DeleteTopic(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, first_group_name);
ForgetTrackedConsumerGroup(stream_name, topic_name, second_group_name);
- ASSERT_THROW(client->get_consumer_groups(make_string_identifier(stream_name), make_string_identifier(topic_name)),
- std::exception);
+ ASSERT_THROW(client.GetConsumerGroups(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupBeforeLoginThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupBeforeLoginThrows) {
RecordProperty("description", "Rejects get_consumer_group before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *setup_client = GetLoggedInClient();
- ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ auto setup_client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(setup_client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- setup_client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
TrackConsumerGroup(stream_name, topic_name, group_name);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
-
- ASSERT_THROW(unauthenticated_client->get_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->get_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->get_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupOnNonExistentResourcesThrows) {
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(unauthenticated_client.GetConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.GetConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.GetConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+}
+
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupOnNonExistentResourcesThrows) {
RecordProperty("description", "Rejects get_consumer_group for streams, topics, or groups that do not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -938,110 +942,110 @@ TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupOnNonExistentResourcesThrows)
const std::string missing_stream_name = GetRandomName();
const std::string missing_topic_name = GetRandomName();
const std::string missing_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), created_group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), created_group_name));
TrackConsumerGroup(stream_name, topic_name, created_group_name);
ASSERT_THROW(
- client->get_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
+ client.GetConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
ASSERT_THROW(
- client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
- ASSERT_THROW(client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(missing_group_name)),
- std::exception);
+ client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
+ ASSERT_THROW(client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(missing_group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, GetConsumerGroupAfterStreamDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, GetConsumerGroupAfterStreamDeletionThrows) {
RecordProperty("description", "Rejects get_consumer_group after deleting the stream that owned the group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
- ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
- ASSERT_THROW(client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupSucceeds) {
+TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupSucceeds) {
RecordProperty("description", "Deletes an existing consumer group successfully.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->delete_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
+ ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_THROW(client->get_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.GetConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupBeforeLoginThrows) {
+TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupBeforeLoginThrows) {
RecordProperty("description", "Rejects delete_consumer_group before connect, and after connect but before login.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *setup_client = GetLoggedInClient();
- ASSERT_NO_THROW(setup_client->create_stream(stream_name));
+ auto setup_client = GetLoggedInHighLevelClient();
+ ASSERT_NO_THROW(setup_client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(
- setup_client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(setup_client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(setup_client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(setup_client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
TrackConsumerGroup(stream_name, topic_name, group_name);
- iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
-
- ASSERT_THROW(unauthenticated_client->delete_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->connect());
- ASSERT_THROW(unauthenticated_client->delete_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
- ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
- ASSERT_NO_THROW(unauthenticated_client->disconnect());
- ASSERT_THROW(unauthenticated_client->delete_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
-}
-
-TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupOnNonExistentResourcesThrows) {
+ auto unauthenticated_client = GetLoggedOutHighLevelClient();
+
+ ASSERT_THROW(unauthenticated_client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Connect());
+ ASSERT_THROW(unauthenticated_client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+ ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
+ ASSERT_NO_THROW(unauthenticated_client.Disconnect());
+ ASSERT_THROW(unauthenticated_client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
+}
+
+TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupOnNonExistentResourcesThrows) {
RecordProperty("description", "Rejects delete_consumer_group for streams, topics, or groups that do not exist.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
@@ -1049,102 +1053,1153 @@ TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupOnNonExistentResourcesThrow
const std::string missing_stream_name = GetRandomName();
const std::string missing_topic_name = GetRandomName();
const std::string missing_group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), created_group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), created_group_name));
TrackConsumerGroup(stream_name, topic_name, created_group_name);
ASSERT_THROW(
- client->delete_consumer_group(make_string_identifier(missing_stream_name), make_string_identifier(topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
+ client.DeleteConsumerGroup(iggy::Identifier::String(missing_stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
ASSERT_THROW(
- client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(missing_topic_name),
- make_string_identifier(created_group_name)),
- std::exception);
- ASSERT_THROW(client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(missing_group_name)),
- std::exception);
+ client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(missing_topic_name),
+ iggy::Identifier::String(created_group_name)),
+ iggy::IggyException);
+ ASSERT_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(missing_group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupTwiceThrows) {
+TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupTwiceThrows) {
RecordProperty("description", "Rejects deleting the same consumer group twice.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->delete_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
+ ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
ForgetTrackedConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_THROW(client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupAfterStreamDeletionThrows) {
+TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupAfterStreamDeletionThrows) {
RecordProperty("description",
"Rejects delete_consumer_group after deleting the stream that owned the consumer group.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
+ 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->delete_stream(make_string_identifier(stream_name)));
+ ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
ForgetTrackedStream(stream_name);
ForgetTrackedConsumerGroup(stream_name, topic_name, group_name);
- ASSERT_THROW(client->delete_consumer_group(make_string_identifier(stream_name), make_string_identifier(topic_name),
- make_string_identifier(group_name)),
- std::exception);
+ ASSERT_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name), iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)),
+ iggy::IggyException);
}
-TEST_F(LowLevelE2E_ConsumerGroup, DeleteConsumerGroupAndRecreateWithSameNameSucceeds) {
+TEST_F(E2E_ConsumerGroup, DeleteConsumerGroupAndRecreateWithSameNameSucceeds) {
RecordProperty("description",
"Allows recreating a consumer group with the same name after the previous group is deleted.");
const std::string stream_name = GetRandomName();
const std::string topic_name = GetRandomName();
const std::string group_name = GetRandomName();
- iggy::ffi::Client *client = GetLoggedInClient();
+ auto client = GetLoggedInHighLevelClient();
- ASSERT_NO_THROW(client->create_stream(stream_name));
+ ASSERT_NO_THROW(client.CreateStream(stream_name));
TrackStream(stream_name);
- ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name,
- make_topic_create_options(1, "none", "server_default", 0, "server_default")));
+ ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
+ iggy::TopicCreateOptions().SetPartitionsCount(1)));
- ASSERT_NO_THROW(client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name));
- ASSERT_NO_THROW(client->delete_consumer_group(
- make_string_identifier(stream_name), make_string_identifier(topic_name), make_string_identifier(group_name)));
+ ASSERT_NO_THROW(client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name));
+ ASSERT_NO_THROW(client.DeleteConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name),
+ iggy::Identifier::String(group_name)));
ASSERT_NO_THROW({
- const auto recreated_group = client->create_consumer_group(make_string_identifier(stream_name),
- make_string_identifier(topic_name), group_name);
+ const auto recreated_group = client.CreateConsumerGroup(iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), group_name);
TrackConsumerGroup(stream_name, topic_name, group_name);
// The VSR server mints group ids monotonically; a recreate gets a
// fresh id (the deleted group held 0), unlike the legacy server which
// reused the freed slot.
- ASSERT_GT(recreated_group.id, 0u);
- ASSERT_EQ(recreated_group.name, group_name);
- ASSERT_EQ(recreated_group.members_count, 0u);
- ASSERT_TRUE(recreated_group.members.empty());
+ ASSERT_GT(recreated_group.Id(), 0u);
+ ASSERT_EQ(recreated_group.Name(), group_name);
+ ASSERT_EQ(recreated_group.MembersCount(), 0u);
+ ASSERT_TRUE(recreated_group.Members().empty());
+ });
+}
+
+TEST_F(E2E_ConsumerGroup, StoreGetAndDeleteConsumerOffsetSucceeds) {
+ 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();
+ 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));
+ 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)),
+ 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), 0, offset),
+ 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), 0, 2));
+ 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), 0, 4));
+ 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), 0, 2));
+
+ ASSERT_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), 0, 5),
+ 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), 1, 0),
+ 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), 0, 1));
+ ASSERT_NO_THROW(client.StoreConsumerOffset(consumer, iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), 0, 3));
+
+ 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), 0, 2));
+
+ 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()), 0, 1));
+
+ 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), 0, 1));
+ 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), 0, 2),
+ 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), 1, 2));
+
+ 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), 0, 1));
+
+ 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), 0, 1));
+ ASSERT_NO_THROW(client.StoreConsumerOffset(first_consumer, iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), 1, 2));
+ ASSERT_NO_THROW(client.StoreConsumerOffset(second_consumer, iggy::Identifier::String(stream_name),
+ iggy::Identifier::String(topic_name), 0, 3));
+
+ 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), 0, 1));
+ 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), 0, 1));
+
+ 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()), 0, 1));
+
+ 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..4f05e27e5 100644
--- a/foreign/cpp/tests/unit/unit_tests.cpp
+++ b/foreign/cpp/tests/unit/unit_tests.cpp
@@ -350,8 +350,10 @@ 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 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.
EXPECT_THROW(client.Connect(), iggy::IggyException);
@@ -374,6 +376,35 @@ 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, maximum_partition, 0); });
+ expect_rejection([&] { (void)client.GetConsumerOffset(consumer, stream, topic, maximum_partition); });
+ expect_rejection([&] { client.DeleteConsumerOffset(consumer, stream, topic, maximum_partition); });
}
TEST(AutoLoginKindTest, HasStableDiscriminantsAndZeroInitializedDefault) {
diff --git a/licenserc.toml b/licenserc.toml
index fdcf10f3a..7339eac25 100644
--- a/licenserc.toml
+++ b/licenserc.toml
@@ -80,6 +80,7 @@ excludes = [
".github/config/hawkeye.version",
"**/.python-version",
"**/gradle/wrapper/**",
+ "**/.clang-tidy",
# cargo-php regenerates this file with an extra separator after the header.
"foreign/php/iggy-php.stubs.php",
]