hubcio commented on code in PR #4180:
URL: https://github.com/apache/iggy/pull/4180#discussion_r4049868835
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -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 {
Review Comment:
suggestion: perhaps `IggyMessageIncoming` / `IggyMessageOutgoing` would be
better name?
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -1985,6 +2423,195 @@ 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 offset Message offset to store.
+ * @param partition_id Partition whose offset is stored.
+ * @throws IggyException if an identifier, partition, or offset is invalid;
+ * the resource does not exist; the client is unauthenticated; the
+ * caller lacks permission; or the request fails.
+ */
+ void StoreConsumerOffset(const Consumer &consumer,
+ const Identifier &stream,
+ const Identifier &topic,
+ std::uint64_t offset,
+ std::uint32_t partition_id);
+
+ /**
+ * @brief Retrieves the stored offset for a consumer or consumer group.
+ *
+ * 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.
+ * @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,
Review Comment:
warning: `UINT32_MAX` means 'any partition' in the bridge (`opt_partition`
turns it into `None`), so a read silently returns partition 0 while store and
delete fail with `InvalidIdentifier`. document the sentinel here, or reject it.
##########
foreign/cpp/include/iggy.hpp:
##########
@@ -1985,6 +2423,195 @@ 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 offset Message offset to store.
+ * @param partition_id Partition whose offset is stored.
+ * @throws IggyException if an identifier, partition, or offset is invalid;
+ * the resource does not exist; the client is unauthenticated; the
+ * caller lacks permission; or the request fails.
+ */
+ void StoreConsumerOffset(const Consumer &consumer,
+ const Identifier &stream,
+ const Identifier &topic,
+ std::uint64_t offset,
Review Comment:
warning: `offset` comes before `partition_id` here, but the bridge call
below and `DeleteConsumerOffset` take `partition_id` first. both are integers,
so a swapped call compiles. swap the last two parameters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
