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/29 08:29:23 UTC
(iggy) branch fix/4292-go-group-membership updated: refactor(go): build the offset write routing target in one place
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/4292-go-group-membership
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/fix/4292-go-group-membership by this push:
new b8025dba2 refactor(go): build the offset write routing target in one place
b8025dba2 is described below
commit b8025dba29f1260311c8acbed5fde2461beceff3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Sep 29 10:23:10 2026 +0200
refactor(go): build the offset write routing target in one place
---
foreign/go/client/tcp/tcp_offset_management.go | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
diff --git a/foreign/go/client/tcp/tcp_offset_management.go b/foreign/go/client/tcp/tcp_offset_management.go
index 8d438c53a..bf6b980fa 100644
--- a/foreign/go/client/tcp/tcp_offset_management.go
+++ b/foreign/go/client/tcp/tcp_offset_management.go
@@ -40,8 +40,7 @@ func (c *IggyTcpClient) GetConsumerOffset(ctx context.Context, consumer iggcon.C
}
func (c *IggyTcpClient) StoreConsumerOffset(ctx context.Context, consumer iggcon.Consumer, streamId iggcon.Identifier, topicId iggcon.Identifier, offset uint64, partitionId *uint32) error {
- target := command.GetConsumerOffset{StreamId: streamId, TopicId: topicId, Consumer: consumer, PartitionId: partitionId}
- return c.writeOffset(ctx, &target, &command.StoreConsumerOffsetRequest{
+ return c.writeOffset(ctx, consumer, streamId, topicId, partitionId, &command.StoreConsumerOffsetRequest{
StreamId: streamId,
TopicId: topicId,
Offset: offset,
@@ -51,8 +50,7 @@ func (c *IggyTcpClient) StoreConsumerOffset(ctx context.Context, consumer iggcon
}
func (c *IggyTcpClient) DeleteConsumerOffset(ctx context.Context, consumer iggcon.Consumer, streamId iggcon.Identifier, topicId iggcon.Identifier, partitionId *uint32) error {
- target := command.GetConsumerOffset{StreamId: streamId, TopicId: topicId, Consumer: consumer, PartitionId: partitionId}
- return c.writeOffset(ctx, &target, &command.DeleteConsumerOffset{
+ return c.writeOffset(ctx, consumer, streamId, topicId, partitionId, &command.DeleteConsumerOffset{
Consumer: consumer,
StreamId: streamId,
TopicId: topicId,
@@ -65,7 +63,7 @@ func (c *IggyTcpClient) DeleteConsumerOffset(ctx context.Context, consumer iggco
// would walk the roster and register a new client identity, which is not a
// member of the group, so a group commit would be refused and the membership
// lost.
-func (c *IggyTcpClient) writeOffset(ctx context.Context, target *command.GetConsumerOffset, write command.Command) error {
+func (c *IggyTcpClient) writeOffset(ctx context.Context, consumer iggcon.Consumer, streamId, topicId iggcon.Identifier, partitionId *uint32, write command.Command) error {
if !c.topologyKnown.Load() {
if _, err := c.GetClusterMetadata(ctx); err != nil {
return err
@@ -75,6 +73,7 @@ func (c *IggyTcpClient) writeOffset(ctx context.Context, target *command.GetCons
_, err := c.do(ctx, write)
return err
}
+ target := command.GetConsumerOffset{StreamId: streamId, TopicId: topicId, Consumer: consumer, PartitionId: partitionId}
routing, err := target.MarshalBinary()
if err != nil {
return err