github-actions[bot] commented on code in PR #4313:
URL: https://github.com/apache/iggy/pull/4313#discussion_r4127716095
##########
foreign/go/client/tcp/tcp_consumer_group_management.go:
##########
@@ -113,11 +113,10 @@ func (c *IggyTcpClient) LeaveConsumerGroup(ctx
context.Context, streamId iggcon.
},
GroupId: groupId,
})
- if err != nil {
- return err
- }
+ // The leave may commit after the caller stops waiting. Preserve its
intent
+ // so a later group poll cannot silently join again.
c.groups.markLeft(newGroupKey(streamId, topicId, groupId))
- return nil
+ return err
Review Comment:
warning: `LeaveConsumerGroup` marks the group left for every error,
including errors raised before the frame is written. Later polls then refuse
the auto-join and return `ErrConsumerGroupMemberNotFound`. Mark left only for
outcomes that can follow a write, and leave `ErrNotConnected` and the other
pre-write errors unmarked.
##########
foreign/go/client/tcp/tcp_offset_management.go:
##########
@@ -40,31 +40,50 @@ 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 {
- // TODO(#4292): a group commit for a partition whose primary is not the
- // coordinator goes out on the coordinator session, is refused as not
- // admitted, and sendFrame walks the roster to the primary. That
reconnect
- // registers a new client identity, which is not a member of the group,
so
- // the replayed commit fails with ConsumerGroupPartitionNotOwned and the
- // membership is gone. Route clustered group commits (and deletes) to
the
- // partition primary through the attached consumer session, as
pollPrimary
- // does for auto-commit polls and the Rust SDK's
PollRouter::write_offset
- // does for offset writes.
- _, err := c.do(ctx, &command.StoreConsumerOffsetRequest{
+ target := command.GetConsumerOffset{StreamId: streamId, TopicId:
topicId, Consumer: consumer, PartitionId: partitionId}
Review Comment:
simplification: `StoreConsumerOffset` and `DeleteConsumerOffset` each build
the same `command.GetConsumerOffset` literal as the routing template for
`writeOffset`. Pass the consumer, stream and topic identifiers to `writeOffset`
instead, so it builds the routing body once.
##########
foreign/go/errors/errors_gen.go:
##########
@@ -2162,6 +2414,9 @@ type InvalidOffset struct {
}
func (e InvalidOffset) Error() string {
+ if e == (InvalidOffset{}) {
Review Comment:
nit: The generated zero-value guard prints `?` for `InvalidOffset{Offset:
0}` and any other error whose real fields are all zero, because it tests the
value, not the origin. Return a code-only error type from `FromCode` and keep
the placeholder there.
--
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]
