kvserver: drop excess replicas when lowering ReplicateQueueMaxSize

Previously, the ReplicateQueueMaxSize cluster setting allowed dynamic adjustment
of the replicate queue’s maximum size. However, decreasing this setting did not
properly drop excess replicas. This commit fixes that by removing replicas when
the queue’s max size is lowered.

Author: wenyihu6

pkg/kv/kvserver/queue.go

@@ -547,8 +547,18 @@ func (bq *baseQueue) SetDisabled(disabled bool) {
 // SetMaxSize sets the max size of the queue.
 func (bq *baseQueue) SetMaxSize(maxSize int64) {
 	bq.mu.Lock()
+	defer bq.mu.Unlock()
 	bq.mu.maxSize = maxSize
-	bq.mu.Unlock()
+	// Drop replicas until no longer exceeding the max size. Note: We call
+	// removeLocked to match the behavior of addInternal. In theory, only
+	// removeFromQueueLocked should be triggered in removeLocked, since the item
+	// is in the priority queue, it should not be processing or in the purgatory
+	// queue. To be safe, however, we use removeLocked.
+	for int64(bq.mu.priorityQ.Len()) > maxSize {
+		pqLen := bq.mu.priorityQ.Len()
+		bq.droppedDueToSize.Inc(1)
+		bq.removeLocked(bq.mu.priorityQ.sl[pqLen-1])
+	}
 }
 
 // lockProcessing locks all processing in the baseQueue. It returns

pkg/kv/kvserver/store_rebalancer_test.go

@@ -1885,15 +1885,19 @@ func TestReplicateQueueMaxSize(t *testing.T) {
 	ReplicateQueueMaxSize.Override(ctx, &tc.store.cfg.Settings.SV, 1)
 	replicateQueue := newReplicateQueue(tc.store, a)
 
-	// Function to add a replica and verify queue state
+	// Helper function to add a replica and verify queue state.
+	verify := func(expectedLength int, expectedDropped int64) {
+		require.Equal(t, expectedLength, replicateQueue.Length())
+		require.Equal(t, expectedDropped, replicateQueue.droppedDueToSize.Count())
+		require.Equal(t, expectedDropped, tc.store.metrics.ReplicateQueueDroppedDueToSize.Count())
+	}
+
 	addReplicaAndVerify := func(rangeID roachpb.RangeID, expectedLength int, expectedDropped int64) {
 		r.Desc().RangeID = rangeID
 		enqueued, err := replicateQueue.testingAdd(context.Background(), r, 0.0)
 		require.NoError(t, err)
 		require.True(t, enqueued)
-		require.Equal(t, expectedLength, replicateQueue.Length())
-		require.Equal(t, expectedDropped, replicateQueue.droppedDueToSize.Count())
-		require.Equal(t, expectedDropped, tc.store.metrics.ReplicateQueueDroppedDueToSize.Count())
+		verify(expectedLength, expectedDropped)
 	}
 
 	// First replica should be added.
@@ -1912,4 +1916,15 @@ func TestReplicateQueueMaxSize(t *testing.T) {
 
 	// Add one more to exceed the max size. Should be dropped.
 	addReplicaAndVerify(102 /* rangeID */, 100 /* expectedLength */, 3 /* expectedDropped */)
+
+	// Reset to the same size should not change the queue length.
+	ReplicateQueueMaxSize.Override(ctx, &tc.store.cfg.Settings.SV, 100)
+	verify(100 /* expectedLength */, 3 /* expectedDropped */)
+
+	// Decrease the max size to 10 which should drop 90 replicas.
+	ReplicateQueueMaxSize.Override(ctx, &tc.store.cfg.Settings.SV, 10)
+	verify(10 /* expectedLength */, 93 /* expectedDropped: 3 + 90 */)
+
+	// Should drop another one now that max size is 10.
+	addReplicaAndVerify(103 /* rangeID */, 10 /* expectedLength */, 94 /* expectedDropped: 3 + 90 + 1 */)
 }