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: mixed_version_cdc.go
    path of file: ./repos/cockroach/pkg/cmd/roachtest/tests
    
    the code of the file until where you have to start completion: // Copyright 2022 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 tests

import (
	"context"
	gosql "database/sql"
	"fmt"
	"math/rand"
	"strconv"
	"strings"
	"time"

	"github.com/IBM/sarama"
	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/registry"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/clusterupgrade"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/mixedversion"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/spec"
	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
	"github.com/cockroachdb/cockroach/pkg/roachprod/logger"
	"github.com/cockroachdb/cock
    
func runCDCMixedVersions(ctx context.Context, t test.Test, c cluster.Cluster) {
	tester := newCDCMixedVersionTester(ctx, t, c)

	// NB: We rely on the testing framework to choose a random predecessor
	// to upgrade from.
	mvt := mixedversion.NewTest(
		ctx, t, t.L(), c, tester.crdbNodes,
		// For now, we perform at most 2 upgrades in this test so that the
		// lowest version we start from is 22.2. The reason for this is
		// that, in 22.1, node draining errors were common and led to
		// changefeeds failing. See #106878.
		//
		// TODO(renato): remove this restriction by not failing the test
		// if the changefeed failed due to "node draining" errors while
		// still in 22.1 or older.
		mixedversion.MaxUpgrades(2),
	)

	cleanupKafka := tester.StartKafka(t, c)
	defer cleanupKafka()

	// MuxRangefeed in various forms is available starting from v22.2.
	setMuxRangeFeedEnabled := func(ctx context.Context, l *logger.Logger, r *rand.Rand, h *mixedversion.Helper) error {
		supported, gatewayNodes, err := tester.muxRangeFeedSupported(h)
		if err != nil {
			return err
		}
		if supported {
			coin := r.Int()%2 == 0
			l.PrintfCtx(ctx, "Setting changefeed.mux_rangefeed.enabled=%t ", coin)
			return h.ExecWithGateway(r, gatewayNodes, "SET CLUSTER SETTING changefeed.mux_rangefeed.enabled=$1", coin)
		}
		return nil
	}

	// Rangefeed scheduler available in 23.2
	setRangeFeedSchedulerEnabled := func(ctx context.Context, l *logger.Logger, r *rand.Rand, h *mixedversion.Helper) error {
		supported, gatewayNodes, err := tester.rangefeedSchedulerSupported(h)
		if err != nil {
			return err
		}
		if supported {
			coin := r.Int()%2 == 0
			l.PrintfCtx(ctx, "Setting kv.rangefeed.scheduler.enabled=%t", coin)
			return h.ExecWithGateway(r, gatewayNodes, "SET CLUSTER SETTING kv.rangefeed.scheduler.enabled=$1", coin)
		}
		return nil
	}

	// Register hooks.
	mvt.OnStartup("start changefeed", tester.createChangeFeed)
	mvt.OnStartup("create validator", tester.setupValidator)
	mvt.OnStartup("init workload", tester.initWorkload)

	runWorkloadCmd := tester.runWorkloadCmd(mvt.RNG())
	_ = mvt.BackgroundCommand("run workload", tester.workloadNodes, runWorkloadCmd)
	_ = mvt.BackgroundFunc("run kafka consumer", tester.runKafkaConsumer)

	// NB: mvt.InMixedVersion will run these hooks multiple times at various points during the rolling upgrade, but
	// not when any nodes are offline. This is important because the validator relies on a db connection.
	mvt.InMixedVersion("wait and validate", tester.waitAndValidate)

	// Enable/disable mux rangefeed related settings in mixed version.
	mvt.InMixedVersion("use mux", setMuxRangeFeedEnabled)
	mvt.InMixedVersion("use scheduler", setRangeFeedSchedulerEnabled)

	mvt.AfterUpgradeFinalized("use mux", setMuxRangeFeedEnabled)
	mvt.AfterUpgradeFinalized("use scheduler", setRangeFeedSchedulerEnabled)
	mvt.AfterUpgradeFinalized("wait and validate", tester.waitAndValidate)
	mvt.Run()
}