Merge #144497 #144649 #144815 #144848

144497: changefeedccl: fix mock kafka server shutdown r=andyyang890 a=asg0451

Fix a test timeout due to a batching sink worker
being blocked on Flush after the context was
cancelled.

Fixes: #144213

Release note: None


144649: ui: refactor schedules page to use useSWR hook r=kyle-a-wong a=kyle-a-wong

- Refactors the schedules page to use `useSwrWithClusterId` instead
  of redux sagas to fetch data.
- Removed show and status props from being stored in redux and
  browser session.
- Deleted schedules page story book

Epic: None
Release note: None

--- 
Note: This is a stacked PR, only the last commit needs to be reviewed 

144815: securitytest: regenerate certs r=rafiss a=rafiss

The certs had expired, causing tests to fail.

Fixes #144782
Release note: None

144848: kvclient: add some metrics for buffered writes r=stevendanna a=arulajmani

Gets the ball rolling on some metrics.

Epic: none
Release note: None

Co-authored-by: Miles Frankel <[email protected]>
Co-authored-by: Kyle Wong <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Arul Ajmani <[email protected]>

Author: 5 people

docs/generated/metrics/metrics.html

@@ -1949,8 +1949,12 @@
 <tr><td>APPLICATION</td><td>txn.restarts.txnpush</td><td>Number of restarts due to a transaction push failure</td><td>Restarted Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>txn.restarts.unknown</td><td>Number of restarts due to a unknown reasons</td><td>Restarted Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>txn.restarts.writetooold</td><td>Number of restarts due to a concurrent writer committing first</td><td>Restarted Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>APPLICATION</td><td>txn.rollbacks.async.failed</td><td>Number of KV transaction that failed to send abort asynchronously which is not always retried</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>APPLICATION</td><td>txn.rollbacks.failed</td><td>Number of KV transaction that failed to send final abort</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.rollbacks.async.failed</td><td>Number of KV transactions that failed to send abort asynchronously which is not always retried</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.rollbacks.failed</td><td>Number of KV transactions that failed to send final abort</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.batches.fully_handled</td><td>Number of KV batches that were fully handled by the write buffer (not sent to KV)</td><td>KV Batches</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.disabled_after_buffering</td><td>Number of KV transactions that disabled write buffering after buffering some writes but before an EndTxn request</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.memory_limit_exceeded</td><td>Number of KV transactions that exceeded the write buffering memory limit</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.num_enabled</td><td>Number of KV transactions that enabled buffered writes</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>SERVER</td><td>build.timestamp</td><td>Build information</td><td>Build Time</td><td>GAUGE</td><td>TIMESTAMP_SEC</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>SERVER</td><td>go.scheduler_latency</td><td>Go scheduling latency</td><td>Nanoseconds</td><td>HISTOGRAM</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>SERVER</td><td>log.buffered.messages.dropped</td><td>Count of log messages that are dropped by buffered log sinks. When CRDB attempts to buffer a log message in a buffered log sink whose buffer is already full, it drops the oldest buffered messages to make space for the new message</td><td>Messages</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>

docs/generated/settings/settings-for-tenants.txt

@@ -99,7 +99,7 @@ kv.transaction.max_refresh_spans_bytes	integer	4194304	maximum number of bytes u
 kv.transaction.randomized_anchor_key.enabled	boolean	false	dictates whether a transactions anchor key is randomized or not	application
 kv.transaction.reject_over_max_intents_budget.enabled	boolean	false	if set, transactions that exceed their lock tracking budget (kv.transaction.max_intents_bytes) are rejected instead of having their lock spans imprecisely compressed	application
 kv.transaction.write_buffering.enabled	boolean	false	if enabled, transactional writes are buffered on the client	application
-kv.transaction.write_buffering.max_buffer_size	integer	4194304	if non-zero, defines that maximum size of the buffer that will be used to buffer transactional writes per-transaction	application
+kv.transaction.write_buffering.max_buffer_size	byte size	4.0 MiB	if non-zero, defines that maximum size of the buffer that will be used to buffer transactional writes per-transaction	application
 kv.transaction.write_pipelining.locking_reads.enabled	boolean	true	if enabled, transactional locking reads are pipelined through Raft consensus	application
 kv.transaction.write_pipelining.ranged_writes.enabled	boolean	true	if enabled, transactional ranged writes are pipelined through Raft consensus	application
 kv.transaction.write_pipelining.enabled (alias: kv.transaction.write_pipelining_enabled)	boolean	true	if enabled, transactional writes are pipelined through Raft consensus	application

docs/generated/settings/settings.html

@@ -128,7 +128,7 @@
 <tr><td><div id="setting-kv-transaction-randomized-anchor-key-enabled" class="anchored"><code>kv.transaction.randomized_anchor_key.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>dictates whether a transactions anchor key is randomized or not</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-transaction-reject-over-max-intents-budget-enabled" class="anchored"><code>kv.transaction.reject_over_max_intents_budget.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>if set, transactions that exceed their lock tracking budget (kv.transaction.max_intents_bytes) are rejected instead of having their lock spans imprecisely compressed</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-transaction-write-buffering-enabled" class="anchored"><code>kv.transaction.write_buffering.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>if enabled, transactional writes are buffered on the client</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
-<tr><td><div id="setting-kv-transaction-write-buffering-max-buffer-size" class="anchored"><code>kv.transaction.write_buffering.max_buffer_size</code></div></td><td>integer</td><td><code>4194304</code></td><td>if non-zero, defines that maximum size of the buffer that will be used to buffer transactional writes per-transaction</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-kv-transaction-write-buffering-max-buffer-size" class="anchored"><code>kv.transaction.write_buffering.max_buffer_size</code></div></td><td>byte size</td><td><code>4.0 MiB</code></td><td>if non-zero, defines that maximum size of the buffer that will be used to buffer transactional writes per-transaction</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-transaction-write-pipelining-locking-reads-enabled" class="anchored"><code>kv.transaction.write_pipelining.locking_reads.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if enabled, transactional locking reads are pipelined through Raft consensus</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-transaction-write-pipelining-ranged-writes-enabled" class="anchored"><code>kv.transaction.write_pipelining.ranged_writes.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if enabled, transactional ranged writes are pipelined through Raft consensus</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-transaction-write-pipelining-enabled" class="anchored"><code>kv.transaction.write_pipelining.enabled<br />(alias: kv.transaction.write_pipelining_enabled)</code></div></td><td>boolean</td><td><code>true</code></td><td>if enabled, transactional writes are pipelined through Raft consensus</td><td>Serverless/Dedicated/Self-Hosted</td></tr>

pkg/ccl/changefeedccl/testfeed_test.go

@@ -1876,12 +1876,16 @@ func (s *fakeKafkaSinkV2) Dial() error {
 				})
 			}
 
-			s.feedCh <- &sarama.ProducerMessage{
+			select {
+			case <-ctx.Done():
+				return kgo.ProduceResults{kgo.ProduceResult{Err: ctx.Err()}}
+			case s.feedCh <- &sarama.ProducerMessage{
 				Topic:     m.Topic,
 				Key:       key,
 				Value:     sarama.ByteEncoder(m.Value),
 				Partition: m.Partition,
 				Headers:   headers,
+			}:
 			}
 		}
 		return nil

pkg/kv/kvclient/kvcoord/txn_coord_sender.go

@@ -321,7 +321,10 @@ func (tc *TxnCoordSender) initCommonInterceptors(
 	if ds, ok := tcf.wrapped.(*DistSender); ok {
 		riGen.ds = ds
 	}
-	tc.interceptorAlloc.txnWriteBuffer.st = tcf.st
+	tc.interceptorAlloc.txnWriteBuffer = txnWriteBuffer{
+		st:         tcf.st,
+		txnMetrics: &tc.metrics,
+	}
 	tc.interceptorAlloc.txnPipeliner = txnPipeliner{
 		st:                       tcf.st,
 		riGen:                    riGen,

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -35,7 +35,7 @@ var BufferedWritesEnabled = settings.RegisterBoolSetting(
 	settings.WithPublic,
 )
 
-var bufferedWritesMaxBufferSize = settings.RegisterIntSetting(
+var bufferedWritesMaxBufferSize = settings.RegisterByteSizeSetting(
 	settings.ApplicationLevel,
 	"kv.transaction.write_buffering.max_buffer_size",
 	"if non-zero, defines that maximum size of the "+
@@ -176,7 +176,8 @@ type txnWriteBuffer struct {
 
 	bufferSeek bufferedWrite // re-use while seeking
 
-	wrapped lockedSender
+	wrapped    lockedSender
+	txnMetrics *TxnMetrics
 
 	// testingOverrideCPutEvalFn is used to mock the evaluation function for
 	// conditional puts. Intended only for tests.
@@ -189,6 +190,9 @@ func (twb *txnWriteBuffer) setEnabled(enabled bool) {
 		// to ensure to flush the buffer.
 		twb.flushOnNextBatch = true
 	}
+	if enabled {
+		twb.txnMetrics.TxnWriteBufferEnabled.Inc(1)
+	}
 	twb.enabled = enabled
 }
 
@@ -246,8 +250,10 @@ func (twb *txnWriteBuffer) SendLocked(
 	// and flush the buffer.
 	maxSize := bufferedWritesMaxBufferSize.Get(&twb.st.SV)
 	bufSize := twb.estimateSize(ba) + twb.bufferSize
-	if bufSize > maxSize {
-		// TODO(arul): add some metrics for this case.
+	// NB: if bufferedWritesMaxBufferSize is set to 0 then we effectively disable
+	// any buffer limiting.
+	if maxSize != 0 && bufSize > maxSize {
+		twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Inc(1)
 		log.VEventf(ctx, 2, "flushing buffer because buffer size (%s) exceeds max size (%s)",
 			humanizeutil.IBytes(bufSize),
 			humanizeutil.IBytes(maxSize))
@@ -277,6 +283,7 @@ func (twb *txnWriteBuffer) SendLocked(
 				return nil, pErr
 			}
 		}
+		twb.txnMetrics.TxnWriteBufferFullyHandledBatches.Inc(1)
 		return br, nil
 	}
 
@@ -1256,6 +1263,13 @@ func (twb *txnWriteBuffer) flushBufferAndSendBatch(
 		return twb.wrapped.SendLocked(ctx, ba) // nothing to flush
 	}
 
+	if _, ok := ba.GetArg(kvpb.EndTxn); !ok {
+		// We're flushing the buffer even though the batch doesn't contain an EndTxn
+		// request. That means we buffered some writes and decided to disable write
+		// buffering mid-way through the transaction, thus necessitating this flush.
+		twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Inc(1)
+	}
+
 	// Flush all buffered writes by pre-pending them to the requests being sent
 	// in the batch.
 	reqs := make([]kvpb.RequestUnion, 0, numBuffered+len(ba.Requests))

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
@@ -25,11 +26,13 @@ import (
 )
 
 func makeMockTxnWriteBuffer(st *cluster.Settings) (txnWriteBuffer, *mockLockedSender) {
+	metrics := MakeTxnMetrics(time.Hour)
 	mockSender := &mockLockedSender{}
 	return txnWriteBuffer{
-		enabled: true,
-		wrapped: mockSender,
-		st:      st,
+		enabled:    true,
+		wrapped:    mockSender,
+		txnMetrics: &metrics,
+		st:         st,
 	}, mockSender
 }
 
@@ -109,6 +112,7 @@ func TestTxnWriteBufferBuffersBlindWrites(t *testing.T) {
 	require.Equal(t, br.Responses[0].GetInner(), &kvpb.PutResponse{})
 	require.Equal(t, br.Responses[1].GetInner(), &kvpb.PutResponse{})
 	require.Equal(t, br.Responses[2].GetInner(), &kvpb.DeleteResponse{})
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferFullyHandledBatches.Count())
 
 	// Verify the writes were buffered correctly.
 	expBufferedWrites := []bufferedWrite{
@@ -1450,6 +1454,8 @@ func TestTxnWriteBufferFlushesWhenOverBudget(t *testing.T) {
 	// Even though we flushed the Put, it shouldn't make it back to the response.
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.DeleteResponse{}, br.Responses[0].GetInner())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Count())
 
 	// Ensure the buffer is empty at this point.
 	require.Equal(t, 0, len(twb.testingBufferedWritesAsSlice()))
@@ -1497,6 +1503,9 @@ func TestTxnWriteBufferFlushesWhenOverBudget(t *testing.T) {
 	require.NotNil(t, br)
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.EndTxnResponse{}, br.Responses[0].GetInner())
+	// Sanity check the metrics remain the same (and don't increase).
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Count())
 }
 
 // TestTxnWriteBufferDeleteRange ensures that the txnWriteBuffer correctly
@@ -1563,6 +1572,7 @@ func TestTxnWriteBufferDeleteRange(t *testing.T) {
 	// Even though we flushed some writes, it shouldn't make it back to the response.
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.DeleteRangeResponse{}, br.Responses[0].GetInner())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
 
 	// Ensure the buffer is empty at this point.
 	require.Equal(t, 0, len(twb.testingBufferedWritesAsSlice()))
@@ -1765,6 +1775,7 @@ func TestTxnWriteBufferFlushesAfterDisabling(t *testing.T) {
 	// The buffer should be flushed and disabled.
 	require.False(t, twb.enabled)
 	require.False(t, twb.flushOnNextBatch)
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
 
 	// Both Put and Del should make it to the server in a single bath.
 	require.Equal(t, numCalledBefore+1, mockSender.NumCalled())
@@ -1813,6 +1824,8 @@ func TestTxnWriteBufferFlushesAfterDisabling(t *testing.T) {
 	require.NotNil(t, br)
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.EndTxnResponse{}, br.Responses[0].GetInner())
+	// Sanity check the metric remains the same (and doesn't increase).
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
 }
 
 // TestTxnWriteBufferClearsBufferOnEpochBump tests that the txnWriteBuffer

pkg/kv/kvclient/kvcoord/txn_metrics.go

@@ -55,6 +55,12 @@ type TxnMetrics struct {
 	// End transaction failure counters.
 	RollbacksFailed      *metric.Counter
 	AsyncRollbacksFailed *metric.Counter
+
+	// Write buffering metrics.
+	TxnWriteBufferEnabled                *metric.Counter
+	TxnWriteBufferDisabledAfterBuffering *metric.Counter
+	TxnWriteBufferMemoryLimitExceeded    *metric.Counter
+	TxnWriteBufferFullyHandledBatches    *metric.Counter
 }
 
 var (
@@ -272,16 +278,40 @@ var (
 	}
 	metaRollbacksFailed = metric.Metadata{
 		Name:        "txn.rollbacks.failed",
-		Help:        "Number of KV transaction that failed to send final abort",
+		Help:        "Number of KV transactions that failed to send final abort",
 		Measurement: "KV Transactions",
 		Unit:        metric.Unit_COUNT,
 	}
 	metaAsyncRollbacksFailed = metric.Metadata{
 		Name:        "txn.rollbacks.async.failed",
-		Help:        "Number of KV transaction that failed to send abort asynchronously which is not always retried",
+		Help:        "Number of KV transactions that failed to send abort asynchronously which is not always retried",
+		Measurement: "KV Transactions",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaTxnWriteBufferEnabled = metric.Metadata{
+		Name:        "txn.write_buffering.num_enabled",
+		Help:        "Number of KV transactions that enabled buffered writes",
+		Measurement: "KV Transactions",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaTxnWriteBufferDisabledAfterBuffering = metric.Metadata{
+		Name:        "txn.write_buffering.disabled_after_buffering",
+		Help:        "Number of KV transactions that disabled write buffering after buffering some writes but before an EndTxn request",
+		Measurement: "KV Transactions",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaTxnWriteBufferLimitExceeded = metric.Metadata{
+		Name:        "txn.write_buffering.memory_limit_exceeded",
+		Help:        "Number of KV transactions that exceeded the write buffering memory limit",
 		Measurement: "KV Transactions",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaTxnWriteBufferFullyHandledBatches = metric.Metadata{
+		Name:        "txn.write_buffering.batches.fully_handled",
+		Help:        "Number of KV batches that were fully handled by the write buffer (not sent to KV)",
+		Measurement: "KV Batches",
+		Unit:        metric.Unit_COUNT,
+	}
 )
 
 // MakeTxnMetrics returns a TxnMetrics struct that contains metrics whose
@@ -321,15 +351,19 @@ func MakeTxnMetrics(histogramWindow time.Duration) TxnMetrics {
 			SigFigs:      3,
 			BucketConfig: metric.Count1KBuckets,
 		}),
-		RestartsWriteTooOld:            telemetry.NewCounterWithMetric(metaRestartsWriteTooOld),
-		RestartsSerializable:           telemetry.NewCounterWithMetric(metaRestartsSerializable),
-		RestartsAsyncWriteFailure:      telemetry.NewCounterWithMetric(metaRestartsAsyncWriteFailure),
-		RestartsCommitDeadlineExceeded: telemetry.NewCounterWithMetric(metaRestartsCommitDeadlineExceeded),
-		RestartsReadWithinUncertainty:  telemetry.NewCounterWithMetric(metaRestartsReadWithinUncertainty),
-		RestartsTxnAborted:             telemetry.NewCounterWithMetric(metaRestartsTxnAborted),
-		RestartsTxnPush:                telemetry.NewCounterWithMetric(metaRestartsTxnPush),
-		RestartsUnknown:                telemetry.NewCounterWithMetric(metaRestartsUnknown),
-		RollbacksFailed:                metric.NewCounter(metaRollbacksFailed),
-		AsyncRollbacksFailed:           metric.NewCounter(metaAsyncRollbacksFailed),
+		RestartsWriteTooOld:                  telemetry.NewCounterWithMetric(metaRestartsWriteTooOld),
+		RestartsSerializable:                 telemetry.NewCounterWithMetric(metaRestartsSerializable),
+		RestartsAsyncWriteFailure:            telemetry.NewCounterWithMetric(metaRestartsAsyncWriteFailure),
+		RestartsCommitDeadlineExceeded:       telemetry.NewCounterWithMetric(metaRestartsCommitDeadlineExceeded),
+		RestartsReadWithinUncertainty:        telemetry.NewCounterWithMetric(metaRestartsReadWithinUncertainty),
+		RestartsTxnAborted:                   telemetry.NewCounterWithMetric(metaRestartsTxnAborted),
+		RestartsTxnPush:                      telemetry.NewCounterWithMetric(metaRestartsTxnPush),
+		RestartsUnknown:                      telemetry.NewCounterWithMetric(metaRestartsUnknown),
+		RollbacksFailed:                      metric.NewCounter(metaRollbacksFailed),
+		AsyncRollbacksFailed:                 metric.NewCounter(metaAsyncRollbacksFailed),
+		TxnWriteBufferEnabled:                metric.NewCounter(metaTxnWriteBufferEnabled),
+		TxnWriteBufferDisabledAfterBuffering: metric.NewCounter(metaTxnWriteBufferDisabledAfterBuffering),
+		TxnWriteBufferMemoryLimitExceeded:    metric.NewCounter(metaTxnWriteBufferLimitExceeded),
+		TxnWriteBufferFullyHandledBatches:    metric.NewCounter(metaTxnWriteBufferFullyHandledBatches),
 	}
 }

pkg/kv/kvclient/kvcoord/txn_test.go

@@ -1659,7 +1659,16 @@ func TestTxnBasicBufferedWrites(t *testing.T) {
 		require.NoError(t, txn.Commit(ctx))
 
 		err := s.DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+			metrics := txn.Sender().(*kvcoord.TxnCoordSender).Metrics()
+			before := metrics.TxnWriteBufferEnabled.Count()
 			txn.SetBufferedWritesEnabled(true)
+			after := metrics.TxnWriteBufferEnabled.Count()
+			require.Equal(t, after-before, int64(1))
+
+			// Ensure that enabling buffered writes after we've already done so
+			// doesn't increment the metric.
+			txn.SetBufferedWritesEnabled(true)
+			require.Equal(t, after, metrics.TxnWriteBufferEnabled.Count())
 
 			// Put transactional value.
 			if err := txn.Put(ctx, keyA, value1); err != nil {