You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: rangefeed_external_test.go
    path of file: ./repos/cockroach/pkg/kv/kvclient/rangefeed
    
    the code of the file until where you have to start completion: // Copyright 2020 The Cockroach Authors.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.

package rangefeed_test

import (
	"context"
	"runtime/pprof"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/cockroachdb/cockroach/pkg/base"
	"github.com/cockroachdb/cockroach/pkg/keys"
	"github.com/cockroachdb/cockroach/pkg/kv"
	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/rangefeed"
	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts"
	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
	"github.com/cockroachdb/cockroach/pkg/roachpb"
	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
	"github.com/cockroachdb/cockroach/pkg/spanconfig"
	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigptsreader"
	"github.com/cockroachdb/cockroach/pkg/storage"
	"github.com/cockroachdb/cockroach/pkg/testutils"
	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
	"github.com/cockroachdb/cockroach/pkg/testutils/storageutils"
	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
	"github.com/cockroachdb/cockroach/pkg/util"
	"github.com/cockroachdb/cockroach/pkg/util/encoding"
	"github.com/cockroachdb/cockroach/pkg/util/future"
	"github.com/cockroachdb/cockroach/pkg/util/hlc"
	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
	"github.com/cockroachdb/cockroach/pkg/util/log"
	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
	"github.com/cockroachdb/errors"
	"github.com/stretchr/testify/assert"
	"github.com/st
    
func TestWithOnFrontierAdvance(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer log.Scope(t).Close(t)

	ctx := context.Background()
	settings := cluster.MakeTestingClusterSettings()
	tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
		ReplicationMode: base.ReplicationManual,
		ServerArgs:      base.TestServerArgs{Settings: settings},
	})
	defer tc.Stopper().Stop(ctx)

	srv := tc.Server(0)
	ts := srv.ApplicationLayer()
	db := ts.DB()
	scratchKey := append(ts.Codec().TenantPrefix(), keys.ScratchRangeMin...)
	_, _, err := srv.SplitRange(scratchKey)
	require.NoError(t, err)
	scratchKey = scratchKey[:len(scratchKey):len(scratchKey)]
	mkKey := func(k string) roachpb.Key {
		return encoding.EncodeStringAscending(scratchKey, k)
	}

	sp := roachpb.Span{
		Key:    scratchKey,
		EndKey: scratchKey.PrefixEnd(),
	}

	for _, l := range []serverutils.ApplicationLayerInterface{ts, srv.SystemLayer()} {
		// Enable rangefeeds, otherwise the thing will retry until they are enabled.
		kvserver.RangefeedEnabled.Override(ctx, &l.ClusterSettings().SV, true)
		// Lower the closed timestamp target duration to speed up the test.
		closedts.TargetDuration.Override(ctx, &l.ClusterSettings().SV, 100*time.Millisecond)
	}

	// Split the range into two so we know the frontier has more than one span to
	// track for certain. We later write to both these ranges.
	_, _, err = srv.SplitRange(mkKey("b"))
	require.NoError(t, err)

	f, err := rangefeed.NewFactory(ts.AppStopper(), db, ts.ClusterSettings(), nil)
	require.NoError(t, err)

	// mu protects secondWriteTS.
	var mu syncutil.Mutex
	secondWriteFinished := false
	frontierAdvancedAfterSecondWrite := false

	// Track the checkpoint TS for spans belonging to both the ranges we split
	// above. This can then be used to compute the minimum timestamp for both
	// these spans. We use the key we write to for the ranges below as keys for
	// this map.
	spanCheckpointTimestamps := make(map[string]hlc.Timestamp)
	forwardCheckpointForKey := func(key string, checkpoint *kvpb.RangeFeedCheckpoint) {
		ts := hlc.MinTimestamp
		if prevTS, found := spanCheckpointTimestamps[key]; found {
			ts = prevTS
		}
		ts.Forward(checkpoint.ResolvedTS)
		spanCheckpointTimestamps[key] = ts
	}
	rows := make(chan *kvpb.RangeFeedValue)
	r, err := f.RangeFeed(ctx, "test", []roachpb.Span{sp}, db.Clock().Now(),
		func(ctx context.Context, value *kvpb.RangeFeedValue) {
			select {
			case rows <- value:
			case <-ctx.Done():
			}
		},
		rangefeed.WithOnCheckpoint(func(ctx context.Context, checkpoint *kvpb.RangeFeedCheckpoint) {
			if checkpoint.Span.ContainsKey(mkKey("a")) {
				forwardCheckpointForKey("a", checkpoint)
			}
			if checkpoint.Span.ContainsKey(mkKey("c")) {
				forwardCheckpointForKey("c", checkpoint)
			}
		}),
		rangefeed.WithOnFrontierAdvance(func(ctx context.Context, frontierTS hlc.Timestamp) {
			minTS := hlc.MaxTimestamp
			for _, ts := range spanCheckpointTimestamps {
				minTS.Backward(ts)
			}
			assert.Truef(
				t,
				frontierTS.Equal(minTS),
				"expected frontier timestamp to be equal to minimum timestamp across spans %s, found %s",
				minTS,
				frontierTS,
			)
			mu.Lock()
			defer mu.Unlock()
			if secondWriteFinished {
				frontierAdvancedAfterSecondWrite = true
			}
		}),
	)
	require.NoError(t, err)
	defer r.Close()

	// Write to a key on both the ranges.
	require.NoError(t, db.Put(ctx, mkKey("a"), 1))

	v := <-rows
	require.Equal(t, mkKey("a"), v.Key)

	require.NoError(t, db.Put(ctx, mkKey("c"), 1))
	mu.Lock()
	secondWriteFinished = true
	mu.Unlock()

	v = <-rows
	require.Equal(t, mkKey("c"), v.Key)

	testutils.SucceedsSoon(t, func() error {
		mu.Lock()
		defer mu.Unlock()
		if frontierAdvancedAfterSecondWrite {
			return nil
		}
		return errors.New("expected frontier to advance after second write")
	})
}