consumer group bugfix: map needs to be one block lower
topicOffsets is meant to track per-topic offsets when updating
uncommitted. Thus, we want to have the maps existence be per-topic.
The prior location had the map per fetch, and would reuse that map
across all topics within a fetch.
@@ -1433,13 +1433,13 @@ func (g *groupConsumer) updateUncommitted(fetches Fetches) {
defer g.mu.Unlock()
for _, fetch := range fetches {
- var topicOffsets map[int32]uncommit
for _, topic := range fetch.Topics {
if g.cl.cfg.logger.Level() >= LogLevelDebug {
fmt.Fprintf(&b, "%s[", topic.Topic)
}
+ var topicOffsets map[int32]uncommit
for _, partition := range topic.Partitions {
if len(partition.Records) == 0 {
continue
