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: parallel_unordered_synchronizer_test.go
    path of file: ./repos/cockroach/pkg/sql/colexec
    
    the code of the file until where you have to start completion: // Copyright 2019 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 colexec

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/cockroachdb/cockroach/pkg/col/coldata"
	"github.com/cockroachdb/cockroach/pkg/col/coldatatestutils"
	"github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecargs"
	"github.com/cockroachdb/cockroach/pkg/sql/colexec/colexectestutils"
	"github.com/cockroachdb/cockroach/pkg/sql/colexecerror"
	"github.com/cockroachdb/cockroach/pkg/sql/colexecop"
	"github.com/cockroachdb/cockroach/pkg/sql/colmem"
	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
	"github.com/cockroachdb/cockroach/pkg/sql/types"
	"github.com/cockroachdb/cockroach/pkg/testutils"
	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
	"github.com/cockroachdb/cockroach/pkg/util/log"
	"github.com/cockroachdb/cockroach/pkg/util/randutil"
	"github.com/cockroachdb/errors"
	"github.com/stretchr/testify/require"
)

func TestParallelUnorderedSynchronizer(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer log.Scope(t).Close(t)

	const (
		maxInputs  = 16
		maxBatches = 16
	)

	var (
		rng, _ = randutil.NewTestRand()
		typs   = []*types.T{types.Int}
		// We want at least two inputs (one regular and one possibly blocking).
		numInputs     = rng.Intn(maxInputs-1) + 2
		numBatches    = rng.Intn(maxBatches) + 1
		blockDuration = time.Second
		errSuffix     = "test-induced metadata"
	)

	type synchronizerTerminationScenario int
	const (
		// synchronizerGracefulTermination is a termination scenario where the
		// synchronizer terminates gracefully.
		synchronizerGracefulTermination synchronizerTerminationScenario = iota
		// synchronizerContextCanceled is a termination scenario where a
		// cancellation requests that a synchronizer terminates.
		synchronizerContextCanceled
		// synchronizerPrematureDrainMeta is a termination scenario where
		// DrainMeta is called prematurely on the synchronizer. In this scenario
		// we also ensure that the inputs are eagerly unblocked from their
		// current Next work.
		synchronizerPrematureDrainMeta
		// synchronizerMaxTerminationScenario should be at the end of the
		// termination scenario list so that it can be used as an upper bound to
		// generate any other termination scenario.
		synchronizerMaxTerminationScenario
	)
	terminationScenario := synchronizerTerminationScenario(rng.Intn(int(synchronizerMaxTerminationScenario)))

	inputs := make([]colexecargs.OpWithMetaInfo, numInputs)
	for i := range inputs {
		var source colexecop.Operator
		args := coldatatestutils.RandomVecArgs{Rand: rng, NullProbability: rng.Float64()}
		batch := coldatatestutils.RandomBatch(testAllocator, args, typs, coldata.BatchSize(), 0 /* length */)
		if i < numInputs-1 {
			s := colexecop.NewRepeatableBatchSource(testAllocator, batch, typs)
			s.ResetBatchesToReturn(numBatches)
			source = s

    
func TestParallelUnorderedSynchronizer(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer log.Scope(t).Close(t)

	const (
		maxInputs  = 16
		maxBatches = 16
	)

	var (
		rng, _ = randutil.NewTestRand()
		typs   = []*types.T{types.Int}
		// We want at least two inputs (one regular and one possibly blocking).
		numInputs     = rng.Intn(maxInputs-1) + 2
		numBatches    = rng.Intn(maxBatches) + 1
		blockDuration = time.Second
		errSuffix     = "test-induced metadata"
	)

	type synchronizerTerminationScenario int
	const (
		// synchronizerGracefulTermination is a termination scenario where the
		// synchronizer terminates gracefully.
		synchronizerGracefulTermination synchronizerTerminationScenario = iota
		// synchronizerContextCanceled is a termination scenario where a
		// cancellation requests that a synchronizer terminates.
		synchronizerContextCanceled
		// synchronizerPrematureDrainMeta is a termination scenario where
		// DrainMeta is called prematurely on the synchronizer. In this scenario
		// we also ensure that the inputs are eagerly unblocked from their
		// current Next work.
		synchronizerPrematureDrainMeta
		// synchronizerMaxTerminationScenario should be at the end of the
		// termination scenario list so that it can be used as an upper bound to
		// generate any other termination scenario.
		synchronizerMaxTerminationScenario
	)
	terminationScenario := synchronizerTerminationScenario(rng.Intn(int(synchronizerMaxTerminationScenario)))

	inputs := make([]colexecargs.OpWithMetaInfo, numInputs)
	for i := range inputs {
		var source colexecop.Operator
		args := coldatatestutils.RandomVecArgs{Rand: rng, NullProbability: rng.Float64()}
		batch := coldatatestutils.RandomBatch(testAllocator, args, typs, coldata.BatchSize(), 0 /* length */)
		if i < numInputs-1 {
			s := colexecop.NewRepeatableBatchSource(testAllocator, batch, typs)
			s.ResetBatchesToReturn(numBatches)
			source = s
		} else {
			// Use the last input to possibly introduce some delay in Next.
			numBatchesLeft := numBatches
			var inputCtx context.Context
			source = &colexecop.CallbackOperator{
				InitCb: func(ctx context.Context) {
					inputCtx = ctx
				},
				NextCb: func() coldata.Batch {
					if numBatchesLeft == 0 {
						return coldata.ZeroBatch
					}
					if terminationScenario == synchronizerPrematureDrainMeta {
						// In PrematureDrainMeta scenario we want to block the
						// input until the context is canceled in order to
						// simulate that this operator is performing some
						// intensive work that should be stopped immediately
						// once the synchronizer transitions into the draining
						// state.
						select {
						case <-inputCtx.Done():
							colexecerror.ExpectedError(inputCtx.Err())
						case <-time.After(blockDuration):
							colexecerror.InternalError(errors.New("unexpectedly slept for a second"))
						}
					}
					numBatchesLeft--
					return batch
				},
			}
		}
		inputs[i].Root = source
		inputIdx := i
		inputs[i].MetadataSources = []colexecop.MetadataSource{
			colexectestutils.CallbackMetadataSource{DrainMetaCb: func() []execinfrapb.ProducerMetadata {
				return []execinfrapb.ProducerMetadata{{Err: errors.Errorf("input %d %s", inputIdx, errSuffix)}}
			}},
		}
	}

	ctx, cancelFn := context.WithCancel(context.Background())

	var wg sync.WaitGroup
	s := NewParallelUnorderedSynchronizer(&execinfra.FlowCtx{Local: true, Gateway: true}, 0 /* processorID */, testAllocator, inputs, &wg)
	s.Init(ctx)

	t.Run(fmt.Sprintf("numInputs=%d/numBatches=%d/terminationScenario=%d", numInputs, numBatches, terminationScenario), func(t *testing.T) {
		if terminationScenario == synchronizerContextCanceled {
			wg.Add(1)
			sleepTime := time.Duration(rng.Intn(500)) * time.Microsecond
			go func() {
				time.Sleep(sleepTime)
				cancelFn()
				wg.Done()
			}()
		} else {
			// Appease the linter complaining about context leaks.
			defer cancelFn()
		}

		batchesReturned := 0
		expectedBatchesReturned := numInputs * numBatches
		for {
			var b coldata.Batch
			if err := colexecerror.CatchVectorizedRuntimeError(func() { b = s.Next() }); err != nil {
				if terminationScenario == synchronizerContextCanceled {
					require.True(t, testutils.IsError(err, "context canceled"), err)
					break
				} else {
					t.Fatal(err)
				}
			}
			if b.Length() == 0 {
				if terminationScenario == synchronizerGracefulTermination {
					// Successful run, check that all inputs have returned metadata.
					meta := s.DrainMeta()
					require.Equal(t, len(inputs), len(meta), "metadata length mismatch, returned metadata is: %v", meta)
				}
				break
			}
			batchesReturned++
			if terminationScenario == synchronizerPrematureDrainMeta && batchesReturned < expectedBatchesReturned {
				// Call DrainMeta before the input is finished.
				meta := s.DrainMeta()
				// Make sure that all expected metadata is still propagated.
				// Note that if the last input wasn't pre-emptied, then the
				// error will not match.
				require.Equal(t, len(inputs), len(meta), "metadata length mismatch, returned metadata is: %v", meta)
				for _, m := range meta {
					require.True(t, strings.Contains(m.Err.Error(), errSuffix))
				}
			}
		}
		if terminationScenario != synchronizerContextCanceled && terminationScenario != synchronizerPrematureDrainMeta {
			require.Equal(t, expectedBatchesReturned, batchesReturned)
		}
		wg.Wait()
	})
}