From 60e2b3093e9bde96d3e82f01a7e4ba89c7881a6e Mon Sep 17 00:00:00 2001 From: ZayanKhan-12 <108294002+ZayanKhan-12@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:57:09 -0400 Subject: [PATCH] storage: fix fetchConsumer panic when a topic is deleted and recreated concurrently fetchConsumer snapshots the consumer group's topics under the consumer lock, releases it, and then reads broker offsets under the broker lock. Storage requests are spread across workers (SetDeleteTopic and SetBrokerOffset go to a random worker, SetConsumerOffset and FetchConsumer to a hashed one), so a topic can be deleted and recreated between the two phases. The broker view of the topic then disagrees with the consumer snapshot in two ways that panic the process: - a partition ring can exist with no stored offsets yet (deleteTopic removes consumer topics and broker state under separate locks, so a consumer commit that lands in between survives the delete; when the recreated topic's partition list is grown by addBrokerOffset, rings for partitions without received offsets are empty), making partition.BrokerOffsets empty and the read of BrokerOffsets[len(BrokerOffsets)-1] panic with index out of range [-1] - the recreated topic can have fewer partitions than the consumer snapshot, so indexing topicMap[p] panics with index out of range Guard both: skip partitions the broker no longer knows about, and only compute current lag when at least one broker offset is present. The consumer offsets are still returned either way, consistent with how a fully deleted topic is already handled. Both states are covered by regression tests that reproduce the exact panic prior to this fix. Co-Authored-By: Claude Fable 5 --- core/internal/storage/inmemory.go | 8 +++- core/internal/storage/inmemory_test.go | 64 ++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/core/internal/storage/inmemory.go b/core/internal/storage/inmemory.go index 0495941a..c15358d6 100644 --- a/core/internal/storage/inmemory.go +++ b/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 { diff --git a/core/internal/storage/inmemory_test.go b/core/internal/storage/inmemory_test.go index 578086ec..74f67897 100644 --- a/core/internal/storage/inmemory_test.go +++ b/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) {