data race in pbqmanager

[vigith]

==================
WARNING: DATA RACE
Read at 0x00c0000e25d0 by goroutine 32:
  runtime.mapiterinit()
      /opt/hostedtoolcache/go/1.19.2/x64/src/runtime/map.go:815 +0x0
  github.com/numaproj/numaflow/pkg/pbq.(*Manager).ShutDown()
      /home/runner/work/numaflow/numaflow/pkg/pbq/pbqmanager.go:216 +0x126
  github.com/numaproj/numaflow/pkg/reduce/readloop.(*ReadLoop).ShutDown()
      /home/runner/work/numaflow/numaflow/pkg/reduce/readloop/readloop.go:227 +0x1bd
  github.com/numaproj/numaflow/pkg/reduce.(*DataForward).Start()
      /home/runner/work/numaflow/numaflow/pkg/reduce/reduce.go:91 +0x16e
  github.com/numaproj/numaflow/pkg/reduce.TestDataForward_StartWithNoOpWM.func7()
      /home/runner/work/numaflow/numaflow/pkg/reduce/reduce_test.go:209 +0x58

Previous write at 0x00c0000e25d0 by goroutine 29:
  runtime.mapdelete_faststr()
      /opt/hostedtoolcache/go/1.19.2/x64/src/runtime/map_faststr.go:301 +0x0
  github.com/numaproj/numaflow/pkg/pbq.(*Manager).deregister()
      /home/runner/work/numaflow/numaflow/pkg/pbq/pbqmanager.go:256 +0x1b2
  github.com/numaproj/numaflow/pkg/pbq.(*PBQ).GC()
      /home/runner/work/numaflow/numaflow/pkg/pbq/pbq.go:91 +0x144
  github.com/numaproj/numaflow/pkg/reduce/pnf.(*ProcessAndForward).Forward()
      /home/runner/work/numaflow/numaflow/pkg/reduce/pnf/processandforward.go:136 +0x5f3
  github.com/numaproj/numaflow/pkg/reduce/readloop.(*orderedForwarder).forward()
      /home/runner/work/numaflow/numaflow/pkg/reduce/readloop/ordered.go:165 +0xdb9
  github.com/numaproj/numaflow/pkg/reduce/readloop.(*orderedForwarder).startUp.func1()
      /home/runner/work/numaflow/numaflow/pkg/reduce/readloop/ordered.go:67 +0x58

Goroutine 32 (running) created at:
  github.com/numaproj/numaflow/pkg/reduce.TestDataForward_StartWithNoOpWM()
      /home/runner/work/numaflow/numaflow/pkg/reduce/reduce_test.go:209 +0xaa9
  testing.tRunner()
      /opt/hostedtoolcache/go/1.19.2/x64/src/testing/testing.go:1446 +0x216
  testing.(*T).Run.func1()
      /opt/hostedtoolcache/go/1.19.2/x64/src/testing/testing.go:1493 +0x47

Goroutine 29 (running) created at:
  github.com/numaproj/numaflow/pkg/reduce/readloop.(*orderedForwarder).startUp()
      /home/runner/work/numaflow/numaflow/pkg/reduce/readloop/ordered.go:67 +0xdc
  github.com/numaproj/numaflow/pkg/reduce/readloop.NewReadLoop()
      /home/runner/work/numaflow/numaflow/pkg/reduce/readloop/readloop.go:84 +0x2de
  github.com/numaproj/numaflow/pkg/reduce.NewDataForward()
      /home/runner/work/numaflow/numaflow/pkg/reduce/reduce.go:70 +0x259
  github.com/numaproj/numaflow/pkg/reduce.TestDataForward_StartWithNoOpWM()
      /home/runner/work/numaflow/numaflow/pkg/reduce/reduce_test.go:206 +0x9ac
  testing.tRunner()
      /opt/hostedtoolcache/go/1.19.2/x64/src/testing/testing.go:1446 +0x216
  testing.(*T).Run.func1()
      /opt/hostedtoolcache/go/1.19.2/x64/src/testing/testing.go:1493 +0x47
==================