diff --git a/core/internal/storage/inmemory.go b/core/internal/storage/inmemory.go index 0495941a..b1996ced 100644 --- a/core/internal/storage/inmemory.go +++ b/core/internal/storage/inmemory.go @@ -858,6 +858,11 @@ func (module *InMemoryStorage) fetchConsumer(request *protocol.StorageRequest, r } for p, partition := range partitions { + if p >= len(topicMap) { + // The topic may have been recreated with fewer partitions + break + } + // Build the slice of broker offsets to return partition.BrokerOffsets = make([]int64, 0, module.intervals) brokerOffsetPtr := topicMap[p].Next() @@ -867,7 +872,7 @@ func (module *InMemoryStorage) fetchConsumer(request *protocol.StorageRequest, r } }) - if len(partition.Offsets) > 0 { + if len(partition.BrokerOffsets) > 0 && len(partition.Offsets) > 0 { brokerOffset := partition.BrokerOffsets[len(partition.BrokerOffsets)-1] lastOffset := partition.Offsets[len(partition.Offsets)-1] if lastOffset != nil { diff --git a/core/internal/storage/inmemory_test.go b/core/internal/storage/inmemory_test.go index 578086ec..4669964c 100644 --- a/core/internal/storage/inmemory_test.go +++ b/core/internal/storage/inmemory_test.go @@ -1135,6 +1135,66 @@ func TestInMemoryStorage_fetchConsumer_BadGroup(t *testing.T) { assert.False(t, ok, "Expected channel to be closed") } +func TestInMemoryStorage_fetchConsumer_RecreatedTopicEmptyRing(t *testing.T) { + startTime := (time.Now().Unix() * 1000) - 100000 + module := startWithTestConsumerOffsets("", startTime) + + // Simulate a topic delete + recreate that left a partition ring with no offsets yet + clusterMap := module.offsets["testcluster"] + clusterMap.brokerLock.Lock() + clusterMap.broker["testtopic"][0] = ring.New(module.intervals) + clusterMap.brokerLock.Unlock() + + request := protocol.StorageRequest{ + RequestType: protocol.StorageFetchConsumer, + Cluster: "testcluster", + Group: "testgroup", + Reply: make(chan interface{}), + } + + // Can't read a reply without concurrency + go module.fetchConsumer(&request, module.Log) + response := <-request.Reply + + assert.IsType(t, protocol.ConsumerTopics{}, response, "Expected response to be of type protocol.ConsumerTopics") + val := response.(protocol.ConsumerTopics) + assert.Len(t, val["testtopic"], 1, "One partition for topic not returned") + assert.Len(t, val["testtopic"][0].BrokerOffsets, 0, "Expected no broker offsets for the recreated partition") + + _, ok := <-request.Reply + assert.False(t, ok, "Expected channel to be closed") +} + +func TestInMemoryStorage_fetchConsumer_RecreatedTopicFewerPartitions(t *testing.T) { + startTime := (time.Now().Unix() * 1000) - 100000 + module := startWithTestConsumerOffsets("", startTime) + + // Simulate a topic delete + recreate with fewer partitions than the consumer has + clusterMap := module.offsets["testcluster"] + clusterMap.brokerLock.Lock() + clusterMap.broker["testtopic"] = make([]*ring.Ring, 0) + clusterMap.brokerLock.Unlock() + + request := protocol.StorageRequest{ + RequestType: protocol.StorageFetchConsumer, + Cluster: "testcluster", + Group: "testgroup", + Reply: make(chan interface{}), + } + + // Can't read a reply without concurrency + go module.fetchConsumer(&request, module.Log) + response := <-request.Reply + + assert.IsType(t, protocol.ConsumerTopics{}, response, "Expected response to be of type protocol.ConsumerTopics") + val := response.(protocol.ConsumerTopics) + assert.Len(t, val["testtopic"], 1, "One partition for topic not returned") + assert.Len(t, val["testtopic"][0].BrokerOffsets, 0, "Expected no broker offsets for the missing partition") + + _, ok := <-request.Reply + assert.False(t, ok, "Expected channel to be closed") +} + func TestInMemoryStorage_fetchConsumer_Expired(t *testing.T) { // We can't insert these offsets normally, so we need to mash them into the module module := startWithTestBrokerOffsets("")