sink: fix concurrent map access in sink manager (#2249) (#2299)

pingcap · Jul 19, 2021 · 578a147 · 578a147
1 parent e3cf5d1
commit 578a147
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 3 deletions.
diff --git a/cdc/sink/manager.go b/cdc/sink/manager.go
@@ -58,6 +58,8 @@ func NewManager(ctx context.Context, backendSink Sink, errCh chan error, checkpo
 
 // CreateTableSink creates a table sink
 func (m *Manager) CreateTableSink(tableID model.TableID, checkpointTs model.Ts) Sink {
+	m.tableSinksMu.Lock()
+	defer m.tableSinksMu.Unlock()
 	if _, exist := m.tableSinks[tableID]; exist {
 		log.Panic("the table sink already exists", zap.Uint64("tableID", uint64(tableID)))
 	}
@@ -67,8 +69,6 @@ func (m *Manager) CreateTableSink(tableID model.TableID, checkpointTs model.Ts)
 		buffer:    make([]*model.RowChangedEvent, 0, 128),
 		emittedTs: checkpointTs,
 	}
-	m.tableSinksMu.Lock()
-	defer m.tableSinksMu.Unlock()
 	m.tableSinks[tableID] = sink
 	return sink
 }

diff --git a/cdc/sink/manager_test.go b/cdc/sink/manager_test.go
@@ -92,8 +92,14 @@ func (s *managerSuite) TestManagerRandom(c *check.C) {
 	var wg sync.WaitGroup
 	tableSinks := make([]Sink, goroutineNum)
 	for i := 0; i < goroutineNum; i++ {
-		tableSinks[i] = manager.CreateTableSink(model.TableID(i), 0)
+		i := i
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			tableSinks[i] = manager.CreateTableSink(model.TableID(i), 0)
+		}()
 	}
+	wg.Wait()
 	for i := 0; i < goroutineNum; i++ {
 		i := i
 		tableSink := tableSinks[i]