storage: fix fetchConsumer panic when a topic is deleted and recreated concurrently

core/internal/storage/inmemory.go

@@ -858,6 +858,12 @@ func (module *InMemoryStorage) fetchConsumer(request *protocol.StorageRequest, r
 		}
 
 		for p, partition := range partitions {
+			if p >= len(topicMap) {
+				// The topic may have been deleted and recreated with fewer partitions since the consumer
+				// snapshot was taken, so the broker may not know about this partition anymore
+				continue
+			}
+
 			// Build the slice of broker offsets to return
 			partition.BrokerOffsets = make([]int64, 0, module.intervals)
 			brokerOffsetPtr := topicMap[p].Next()
@@ -867,7 +873,7 @@ func (module *InMemoryStorage) fetchConsumer(request *protocol.StorageRequest, r
 				}
 			})
 
-			if len(partition.Offsets) > 0 {
+			if len(partition.Offsets) > 0 && len(partition.BrokerOffsets) > 0 {
 				brokerOffset := partition.BrokerOffsets[len(partition.BrokerOffsets)-1]
 				lastOffset := partition.Offsets[len(partition.Offsets)-1]
 				if lastOffset != nil {

core/internal/storage/inmemory_test.go

@@ -1182,6 +1182,70 @@ func TestInMemoryStorage_fetchConsumer_Expired(t *testing.T) {
 	assert.False(t, ok, "Expected channel to be closed")
 }
 
+func TestInMemoryStorage_fetchConsumer_RecreatedTopicNoBrokerOffsets(t *testing.T) {
+	startTime := (time.Now().Unix() * 1000) - 100000
+	module := startWithTestConsumerOffsets("", startTime)
+
+	// Simulate the state left behind when a consumer commit races the two halves of a deleteTopic on
+	// another worker and the topic is then recreated: the broker ring for the partition exists (created
+	// by addBrokerOffset growing the partition list) but holds 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.Equal(t, uint64(0), val["testtopic"][0].CurrentLag, "Expected current lag to be zero when no broker offset is known")
+	assert.Len(t, val["testtopic"][0].Offsets, 10, "Expected consumer offsets to still be returned")
+
+	_, 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 the broker view of the topic having fewer partitions than the consumer snapshot, as
+	// happens when the topic is recreated with fewer partitions while a fetch is in flight
+	clusterMap := module.offsets["testcluster"]
+	clusterMap.brokerLock.Lock()
+	clusterMap.broker["testtopic"] = clusterMap.broker["testtopic"][: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.Equal(t, uint64(0), val["testtopic"][0].CurrentLag, "Expected current lag to be zero when the broker partition is not known")
+
+	_, ok := <-request.Reply
+	assert.False(t, ok, "Expected channel to be closed")
+}
+
 // TODO: Test for clear consumer offsets, including clear for missing group
 
 func TestInMemoryStorage_fetchConsumersForTopic(t *testing.T) {