Merge #153723

153723: kvserver: deflake TestClosedTimestampFrozenAfterSubsumption r=iskettaneh a=iskettaneh

This commit deflakes TestClosedTimestampFrozenAfterSubsumption by making sure
that when running with leader leases, we first make sure that the leader is
fortified before stopping storeliveness heartbeats.

Otherwise, there could be a case where the leader hasn't been fortified yet, and
blocking storeliveness heartbeats won't necessarily cause the leader to step down
as the leader will be sending normal raft heartbeats since it never got fortified
for this term.

Fixes: #153679

Release note: None

Co-authored-by: Ibrahim Kettaneh <ibrahim.kettaneh@cockroachlabs.com>

Author: craig[bot]

pkg/kv/kvserver/closed_timestamp_test.go

@@ -624,8 +624,11 @@ func TestClosedTimestampFrozenAfterSubsumption(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.UnderRace(t)
-	skip.UnderDeadlock(t)
+	skip.UnderDuress(t)
+
+	// Increase the verbosity of the logs to help debug the test if it fails, especially
+	// raft related logs when the test tries to transfer the lease non-cooperatively.
+	require.NoError(t, log.SetVModule("raft=4,*=1"))
 
 	for _, test := range []struct {
 		name string

pkg/testutils/testcluster/testcluster.go

@@ -1267,6 +1267,7 @@ func (tc *TestCluster) MoveRangeLeaseNonCooperatively(
 		return nil, errors.Errorf("must set StoreTestingKnobs.AllowLeaseRequestProposalsWhenNotLeader")
 	}
 
+	log.Dev.Infof(ctx, "moving lease non-cooperatively of range %v to %v", rangeDesc, dest)
 	destServer, err := tc.FindMemberServer(dest.StoreID)
 	if err != nil {
 		return nil, err
@@ -1380,8 +1381,17 @@ func (tc *TestCluster) ensureLeaderStepsDown(
 				leaderStore = curStore
 				leaderNode = s
 				leaderReplica = curR
+
+				// Make sure that the leader is fortified because in the next step we
+				// will stop store liveness messages to the leader, and we want to cause
+				// it to step down. If the leader isn't fortified yet, stopping store
+				// liveness messages to it will not cause it to step down.
+				if curR.RaftStatus().LeadSupportUntil.IsEmpty() {
+					return errors.Errorf("leader is not fortified")
+				}
 			}
 		}
+
 		// At this point we have iterated over all nodes in the cluster, if we
 		// haven't found a leader, wait for a bit for one to step up.
 		if leaderStore == nil {