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: retl_test.go
    path of file: ./repos/rudder-server/integration_test/retl_test
    
    the code of the file until where you have to start completion: package retltest

import (
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
	"github.com/rudderlabs/rudder-server/services/rsources"
)

type rETLSource struct {
	id   string
	once sync.Once
}

func (s *rETLSource) ID() string {
	s.once.Do(func() {
		s.id = rand.String(27)
	})
	return s.id
}

func TestRETL(t *testing.T) {
	rETLSource := &rETLSource{}
	webhook_1 := &webhook{name: "webhook_1"}
	webhook_2 := &webhook{name: "webhook_2"}

	s := &SUT{
		Sources: []srcWithDst{
	
    
func TestRETL(t *testing.T) {
	rETLSource := &rETLSource{}
	webhook_1 := &webhook{name: "webhook_1"}
	webhook_2 := &webhook{name: "webhook_2"}

	s := &SUT{
		Sources: []srcWithDst{
			Connect(rETLSource, webhook_1, webhook_2),
		},
	}
	s.Start(t)
	defer s.Shutdown(t)

	jobID := rand.String(27)
	jobRunID := rand.String(27)

	taskRunID := rand.String(27)
	numOfRecords := 10

	s.SendRETL(t, rETLSource.ID(), webhook_1.ID(), manyRecords(rudderSource{
		JobID:     jobID,
		JobRunID:  jobRunID,
		TaskRunID: taskRunID,
	}, "identify", numOfRecords))

	require.Eventually(t, func() bool {
		return !pendingTask(s.JobStatus(t, rETLSource.ID(), jobRunID, taskRunID))
	}, 15*time.Second, 250*time.Millisecond, "using job-status to check if the task is completed")

	require.Equal(t, numOfRecords, webhook_1.Count())
	require.Zero(t, webhook_2.Count(), "webhook_2 should not receive any events")

	t.Run("second task is sending to webhook_2", func(t *testing.T) {
		taskRunID2 := rand.String(27)
		jobRunID2 := rand.String(27)
		numOfRecords2 := 8

		s.SendRETL(t, rETLSource.ID(), webhook_2.ID(), manyRecords(rudderSource{
			JobID:     jobID,
			JobRunID:  jobRunID2,
			TaskRunID: taskRunID2,
		}, "identify", numOfRecords2))

		require.Eventually(t, func() bool {
			t.Logf("webhook count %d", webhook_2.Count())
			return !pendingTask(s.JobStatus(t, rETLSource.ID(), jobRunID2, taskRunID2))
		}, 10*time.Second, 2*time.Second, "using job-status to check if the task is completed")

		require.Equal(t, numOfRecords, webhook_1.Count())
		require.Equal(t, numOfRecords2, webhook_2.Count())
	})

	// TODO: add test for failed-records and delete failed-records
}