OffsetFetch() doesn't return topic info if topics is unset in request

[hhahn-tw]

Describe the bug

Kafka's documentation for OffsetFetch() indicates that in the OffsetFetchRequest object, the topics field should be set to "Each topic we would like to fetch offsets for, or null to fetch offsets for all topics."

However, when topics is unset in kafka-go (or set explicitly to nil), the response does not return any information on topics for the given consumer group.

In comparison, sarama's equivalent, FetchOffset() returns offset info for all known topics of a given consumer group (even if they are Empty), as does an equivalent python library

Kafka Version

This is tested against kafka 2.4.1 using kafka-go 0.4.39

To Reproduce

Call Client.OffsetFetch() with no topics field set in the OffsetFetchRequest object.

package main

import (
    "github.com/segmentio/kafka-go"
    "context"
)

func main() {
    ctx := context.Background()
    client := kafka.Client{Addr: kafka.TCP("localhost:9092") }
    offsets,_ := client.OffsetFetch(ctx, &kafka.OffsetFetchRequest{ GroupID: "my-group" })
    ...
}

Expected Behavior

All discovered topics and offsets for the requested consumer group in response. For example, sarama returns:

my-group : &sarama.OffsetFetchResponse{
    Version:        2,
    ThrottleTimeMs: 0,
    Blocks:         {
        "my-topic": {
            1:  &sarama.OffsetFetchResponseBlock{Offset:11, LeaderEpoch:0, Metadata:"", Err:0},
            3:  &sarama.OffsetFetchResponseBlock{Offset:10, LeaderEpoch:0, Metadata:"", Err:0},
            2:  &sarama.OffsetFetchResponseBlock{Offset:10, LeaderEpoch:0, Metadata:"", Err:0},
            4:  &sarama.OffsetFetchResponseBlock{Offset:10, LeaderEpoch:0, Metadata:"", Err:0},
            0:  &sarama.OffsetFetchResponseBlock{Offset:11, LeaderEpoch:0, Metadata:"", Err:0},
            5:  &sarama.OffsetFetchResponseBlock{Offset:10, LeaderEpoch:0, Metadata:"", Err:0},
        },
    },
    Err: 0,
}

Observed Behavior

No topic/offset is returned in the response:

my-group: &kafka.OffsetFetchResponse{
    Throttle: 0,
    Topics:   {
    },
    Error: nil,
}

[petedannemann]

I think this issue stems from two things:

1. Topics in the request is not tagged as nullable. When this gets encoded, the tag is used to ensure this is encoded as null bytes
2. We use an empty slice and not a nil slice for the topics array when there are no topics provided. Topics needs to be be a nil array, not an empty array for this to get encoded as null

[petedannemann]

Fixed by #1162