Stop and start pool between tests to prevent data races

This is necessary since the test introspects internal state.

pkg/broker/handler/pool/fanout/pool_test.go

@@ -48,7 +48,8 @@ func TestWatchAndSync(t *testing.T) {
 	targets := make(chan *config.TargetsConfig)
 	defer close(targets)
 	emptyTargetConfig := new(config.TargetsConfig)
-	p, err := StartSyncPool(ctx, config.NewTargetsWatcher(targets),
+	watcher := config.NewTargetsWatcher(targets)
+	p, err := StartSyncPool(ctx, watcher,
 		pool.WithPubsubClient(ps),
 		pool.WithProjectID(testProject),
 	)
@@ -69,9 +70,20 @@ func TestWatchAndSync(t *testing.T) {
 		targets <- targetConfig
 		// Wait a short period for the handlers to be updated.
 		<-time.After(time.Second)
+		cancel()
 		assertHandlers(t, p, targetConfig)
 	})
 
+	ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+	p, err = StartSyncPool(ctx, watcher,
+		pool.WithPubsubClient(ps),
+		pool.WithProjectID(testProject),
+	)
+	if err != nil {
+		t.Errorf("unexpected error from starting sync pool: %v", err)
+	}
+
 	t.Run("adding and deleting brokers changes handlers", func(t *testing.T) {
 		newTargetConfig := &config.TargetsConfig{Brokers: make(map[string]*config.Broker)}
 		// Populate with new brokers.
@@ -83,9 +95,20 @@ func TestWatchAndSync(t *testing.T) {
 		targets <- newTargetConfig
 		// Wait a short period for the handlers to be updated.
 		<-time.After(time.Second)
+		cancel()
 		assertHandlers(t, p, newTargetConfig)
 	})
 
+	ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+	p, err = StartSyncPool(ctx, watcher,
+		pool.WithPubsubClient(ps),
+		pool.WithProjectID(testProject),
+	)
+	if err != nil {
+		t.Errorf("unexpected error from starting sync pool: %v", err)
+	}
+
 	t.Run("deleting all brokers deletes all handlers", func(t *testing.T) {
 		// clean up all brokers
 		targets <- emptyTargetConfig