From 32df246a81972c152fbececa664e04f3a376a33a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 31 Aug 2026 09:44:34 -0700 Subject: [PATCH] mq: fix idle-cleanup shard deadlock that permanently wedges the broker's topic map (#11051) mq: remove emptied topics after the cleanup iteration, not inside it cleanupIdlePartitions called manager.topics.Remove from inside manager.topics.IterCb. IterCb holds the shard's read lock while running the callback, and Remove takes the same shard's write lock, so removing an emptied topic self-deadlocked the cleanup goroutine. The pending writer then blocked every later reader of that shard, permanently hanging ListTopicsInMemory and, for shard-mates, TopicExistsInMemory. On the Kafka gateway this surfaced as flaky e2e consumer-group tests: one minute after any earlier topic went idle, the broker's first 'Removing empty topic' wedged the map, every gateway ListTopics/TopicExists RPC burned its full 5s timeout, Metadata could no longer finish inside kafka-go's 5s coordinator deadline, and consumer groups looped in PreparingRebalance until the test timed out. Collect the emptied topic keys during the iteration and remove them afterwards via RemoveCb, re-checking emptiness under the shard lock so a topic that just gained a partition is kept. Claude-Session: https://claude.ai/code/session_014yA6c8JQcY6MqPXCT13yYA --- weed/mq/topic/local_manager.go | 27 +++++++++-- weed/mq/topic/local_manager_test.go | 75 +++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 3 deletions(-) create mode 100644 weed/mq/topic/local_manager_test.go diff --git a/weed/mq/topic/local_manager.go b/weed/mq/topic/local_manager.go index bc33fdab0..89a2ac8c1 100644 --- a/weed/mq/topic/local_manager.go +++ b/weed/mq/topic/local_manager.go @@ -52,6 +52,7 @@ func (manager *LocalTopicManager) StartIdlePartitionCleanup(ctx context.Context, // cleanupIdlePartitions removes idle partitions from memory func (manager *LocalTopicManager) cleanupIdlePartitions(idleTimeout time.Duration) { cleanedCount := 0 + var emptyTopics []string // Iterate through all topics manager.topics.IterCb(func(topicKey string, localTopic *LocalTopic) { @@ -78,13 +79,33 @@ func (manager *LocalTopicManager) cleanupIdlePartitions(idleTimeout time.Duratio } } - // If topic has no partitions left, remove it if len(localTopic.Partitions) == 0 { - glog.V(1).Infof("Removing empty topic %s", topicKey) - manager.topics.Remove(topicKey) + emptyTopics = append(emptyTopics, topicKey) } }) + // Remove emptied topics only after the iteration: IterCb holds the shard's + // read lock while running the callback, so calling Remove (which takes the + // same shard's write lock) inside it self-deadlocks and permanently wedges + // the shard — every later reader of that shard blocks behind the stuck + // writer, hanging ListTopicsInMemory/TopicExistsInMemory forever. + for _, topicKey := range emptyTopics { + manager.topics.RemoveCb(topicKey, func(key string, localTopic *LocalTopic, exists bool) bool { + if !exists || localTopic == nil { + return false + } + // Re-check emptiness under the shard's write lock: a publisher or + // subscriber may have added a partition since the scan above. + localTopic.partitionLock.RLock() + defer localTopic.partitionLock.RUnlock() + if len(localTopic.Partitions) > 0 { + return false + } + glog.V(1).Infof("Removing empty topic %s", key) + return true + }) + } + if cleanedCount > 0 { glog.V(0).Infof("Cleaned up %d idle partition(s)", cleanedCount) } diff --git a/weed/mq/topic/local_manager_test.go b/weed/mq/topic/local_manager_test.go new file mode 100644 index 000000000..4d0e32091 --- /dev/null +++ b/weed/mq/topic/local_manager_test.go @@ -0,0 +1,75 @@ +package topic + +import ( + "testing" + "time" +) + +// cleanupIdlePartitions used to call manager.topics.Remove from inside +// manager.topics.IterCb. IterCb holds the shard's read lock while running the +// callback, and Remove takes the same shard's write lock, so removing an +// emptied topic deadlocked the cleanup goroutine and permanently wedged the +// shard: every later ListTopicsInMemory / TopicExistsInMemory blocked behind +// the stuck writer. Observed in CI as the Kafka gateway's Metadata requests +// timing out forever once any topic went idle mid-test. +func TestCleanupIdlePartitionsRemovesEmptyTopicWithoutDeadlock(t *testing.T) { + manager := NewLocalTopicManager() + + tp := NewTopic("test", "idle-topic") + localTopic := NewLocalTopic(tp) + localPartition := NewLocalPartition(Partition{RingSize: 2520, RangeStart: 0, RangeStop: 2520}, 1, nil, nil) + localTopic.Partitions = append(localTopic.Partitions, localPartition) + manager.topics.Set(tp.String(), localTopic) + + // Make the partition idle beyond any timeout. + localPartition.lastActivityTime.Store(time.Now().Add(-time.Hour).UnixNano()) + + done := make(chan struct{}) + go func() { + manager.cleanupIdlePartitions(time.Minute) + close(done) + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("cleanupIdlePartitions deadlocked while removing an emptied topic") + } + + if manager.topics.Has(tp.String()) { + t.Errorf("emptied topic %s should have been removed", tp.String()) + } + + // The map must still be fully usable afterwards — with the deadlock the + // wedged shard made this call block forever. + listed := make(chan []Topic, 1) + go func() { listed <- manager.ListTopicsInMemory() }() + select { + case topics := <-listed: + if len(topics) != 0 { + t.Errorf("expected no topics in memory, got %v", topics) + } + case <-time.After(10 * time.Second): + t.Fatal("ListTopicsInMemory blocked after cleanup — shard still wedged") + } +} + +// A topic that gains a partition between the cleanup scan and the removal must +// be kept: removal re-checks emptiness under the shard write lock. +func TestCleanupIdlePartitionsKeepsActiveTopic(t *testing.T) { + manager := NewLocalTopicManager() + + tp := NewTopic("test", "active-topic") + localPartition := NewLocalPartition(Partition{RingSize: 2520, RangeStart: 0, RangeStop: 2520}, 1, nil, nil) + manager.AddLocalPartition(tp, localPartition) + + // Fresh activity: nothing should be cleaned up. + manager.cleanupIdlePartitions(time.Minute) + + if !manager.topics.Has(tp.String()) { + t.Errorf("active topic %s should not have been removed", tp.String()) + } + if manager.GetLocalPartition(tp, localPartition.Partition) == nil { + t.Errorf("active partition should still be registered") + } +}