From ec7857ba94c602dec58d1de1e5cee64deed492dd Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 23 Oct 2026 15:56:12 +0800 Subject: [PATCH 1/2] fix the dispatcher manager stale --- .../dispatchermanager/dispatcher_manager.go | 5 + .../dispatcher_manager_test.go | 4 + .../dispatcher_orchestrator.go | 22 ++- .../dispatcher_orchestrator_test.go | 140 ++++++++++++++++++ 4 files changed, 170 insertions(+), 1 deletion(-) diff --git a/downstreamadapter/dispatchermanager/dispatcher_manager.go b/downstreamadapter/dispatchermanager/dispatcher_manager.go index e2617ec3c9..f801d2c819 100644 --- a/downstreamadapter/dispatchermanager/dispatcher_manager.go +++ b/downstreamadapter/dispatchermanager/dispatcher_manager.go @@ -1054,6 +1054,11 @@ func (e *DispatcherManager) mergeEventDispatcher(dispatcherIDs []common.Dispatch // ==== remove and clean related functions ==== +// IsClosing reports whether shutdown has started, including a completed shutdown. +func (e *DispatcherManager) IsClosing() bool { + return e.closing.Load() || e.closed.Load() +} + func (e *DispatcherManager) TryClose(removeChangefeed bool) bool { if removeChangefeed { e.removeChangefeedRequested.Store(true) diff --git a/downstreamadapter/dispatchermanager/dispatcher_manager_test.go b/downstreamadapter/dispatchermanager/dispatcher_manager_test.go index 0e7ca67f84..c766e6cc36 100644 --- a/downstreamadapter/dispatchermanager/dispatcher_manager_test.go +++ b/downstreamadapter/dispatchermanager/dispatcher_manager_test.go @@ -518,6 +518,7 @@ func TestTryCloseRemovedRequestAfterClosedReturnsImmediatelyAndTriggersCleanup(t func TestLocalFenceCancelsWritePathWithoutWaitingForCleanup(t *testing.T) { manager := createTestManager(t) + require.False(t, manager.IsClosing()) ctx, cancel := context.WithCancel(context.Background()) manager.ctx = ctx manager.cancel = cancel @@ -535,11 +536,14 @@ func TestLocalFenceCancelsWritePathWithoutWaitingForCleanup(t *testing.T) { require.FailNow(t, "local fence should not wait for dispatcher manager cleanup") } require.ErrorIs(t, ctx.Err(), context.Canceled) + require.True(t, manager.IsClosing()) + require.False(t, manager.TryClose(false)) manager.wg.Done() require.Eventually(t, func() bool { return manager.TryClose(false) }, time.Second, 10*time.Millisecond) + require.True(t, manager.IsClosing()) } func TestLocalFenceDoesNotWaitForBootstrapWriteBlockEvent(t *testing.T) { diff --git a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go index 2696503cb2..bb1a0715bb 100644 --- a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go +++ b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go @@ -197,6 +197,23 @@ func (m *DispatcherOrchestrator) handleBootstrapRequest( closedEpoch, closed := m.closedMaintainerEpochs[cfId] m.mutex.Unlock() + if exists && manager.IsClosing() { + manager.MaintainerFenceMu.Lock() + if !manager.CanUpdateMaintainer(from, maintainerEpoch) || !manager.TryClose(false) { + manager.MaintainerFenceMu.Unlock() + // Leave bootstrap pending until the old manager has released its resources. + return nil + } + m.mutex.Lock() + delete(m.dispatcherManagers, cfId) + m.recordClosedMaintainerEpochLocked(cfId, manager.GetMaintainerEpoch(), false) + closedEpoch, closed = m.closedMaintainerEpochs[cfId] + m.mutex.Unlock() + manager.MaintainerFenceMu.Unlock() + metrics.DispatcherManagerGauge.WithLabelValues(cfId.Keyspace(), cfId.Name()).Dec() + exists = false + } + var err error if !exists { if closed && (maintainerEpoch == 0 || maintainerEpoch <= closedEpoch) { @@ -668,10 +685,13 @@ func (m *DispatcherOrchestrator) handleCloseRequest( decGauge := false manager.MaintainerFenceMu.Lock() if manager.IsMaintainerRequestAllowed(from, req.MaintainerEpoch) { + // Preserve removal tombstones if bootstrap later retires this manager. + m.mutex.Lock() + m.recordClosedMaintainerEpochLocked(cfId, req.MaintainerEpoch, req.Removed) + m.mutex.Unlock() if closed := manager.TryClose(req.Removed); closed { m.mutex.Lock() delete(m.dispatcherManagers, cfId) - m.recordClosedMaintainerEpochLocked(cfId, req.MaintainerEpoch, req.Removed) m.mutex.Unlock() decGauge = true response.Success = true diff --git a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go index 2e0d205f22..06e47096ae 100644 --- a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go +++ b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go @@ -844,6 +844,146 @@ func TestHandleCloseRequestAcksStaleMaintainerEpoch(t *testing.T) { require.NotContains(t, orchestrator.closedMaintainerEpochs, cfID) } +func TestHandleBootstrapRequestCanCreateDispatcherAfterHandover(t *testing.T) { + for _, tc := range []struct { + state string + oldEpoch uint64 + newEpoch uint64 + }{ + {"running", 2, 3}, + {"closing", 2, 3}, + {"closed", 2, 3}, + {"closed", 0, 0}, + } { + t.Run(fmt.Sprintf("%s epoch %d to %d", tc.state, tc.oldEpoch, tc.newEpoch), func(t *testing.T) { + mc := messaging.NewMockMessageCenter() + setupBootstrapTestServices(t, mc, true) + + cfID := common.NewChangeFeedIDWithName("cf", "default") + configBytes, err := json.Marshal(newBootstrapResponseTestChangefeedConfig(cfID)) + require.NoError(t, err) + orchestrator := newTestDispatcherOrchestrator() + orchestrator.mc = mc + req := &heartbeatpb.MaintainerBootstrapRequest{ + ChangefeedID: cfID.ToPB(), + Config: configBytes, + StartTs: 100, + MaintainerEpoch: tc.oldEpoch, + } + require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("old-maintainer"), req)) + requireMessage(t, mc.GetMessageChannel(), "initial bootstrap response was not sent") + manager := orchestrator.dispatcherManagers[cfID] + require.NotNil(t, manager) + cleanupDispatcherManager(t, manager) + if tc.state != "running" { + require.NoError(t, orchestrator.handleCloseRequest(node.ID("old-maintainer"), &heartbeatpb.MaintainerCloseRequest{ + ChangefeedID: cfID.ToPB(), + MaintainerEpoch: tc.oldEpoch, + })) + msg := requireMessage(t, mc.GetMessageChannel(), "close response was not sent") + require.False(t, msg.Message[0].(*heartbeatpb.MaintainerCloseResponse).Success) + + // The old maintainer dies before retrying close. Local cleanup finishes, + // but no further close request removes the manager from the orchestrator. + if tc.state == "closed" { + require.Eventually(t, func() bool { + return manager.TryClose(false) + }, time.Second, 10*time.Millisecond) + } + require.Same(t, manager, orchestrator.dispatcherManagers[cfID]) + } + + req.MaintainerEpoch = tc.newEpoch + var response *heartbeatpb.MaintainerBootstrapResponse + bootstrap := func() bool { + require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("new-maintainer"), req)) + select { + case msg := <-mc.GetMessageChannel(): + response = msg.Message[0].(*heartbeatpb.MaintainerBootstrapResponse) + return true + default: + require.Same(t, manager, orchestrator.dispatcherManagers[cfID]) + require.Equal(t, node.ID("old-maintainer"), manager.GetMaintainerID()) + return false + } + } + if !bootstrap() { + require.Eventually(t, bootstrap, time.Second, 10*time.Millisecond) + } + recovered := orchestrator.dispatcherManagers[cfID] + require.NotNil(t, recovered) + if recovered != manager { + cleanupDispatcherManager(t, recovered) + } + if tc.state == "running" { + require.Same(t, manager, recovered) + } else { + require.NotSame(t, manager, recovered) + } + require.Nil(t, response.Err) + require.Equal(t, tc.newEpoch, response.MaintainerEpoch) + require.Equal(t, node.ID("new-maintainer"), recovered.GetMaintainerID()) + + // A successful handover must leave a manager capable of creating dispatchers. + err = recovered.NewTableTriggerEventDispatcher(common.NewDispatcherID().ToPB(), 100, false) + require.NoError(t, err, "bootstrap succeeded but the dispatcher manager write path is still closed") + }) + } +} + +func TestHandleBootstrapRequestRejectsStaleEpochAfterManagerClose(t *testing.T) { + for _, tc := range []struct { + oldEpoch uint64 + requestEpoch uint64 + removed bool + }{ + {2, 0, false}, + {2, 1, false}, + {2, 2, false}, + {0, 0, true}, + } { + t.Run(fmt.Sprintf("epoch %d to %d removed %t", tc.oldEpoch, tc.requestEpoch, tc.removed), func(t *testing.T) { + mc := messaging.NewMockMessageCenter() + setupBootstrapTestServices(t, mc, false) + cfID := common.NewChangeFeedIDWithName("cf", "default") + configBytes, err := json.Marshal(newBootstrapResponseTestChangefeedConfig(cfID)) + require.NoError(t, err) + orchestrator := newTestDispatcherOrchestrator() + orchestrator.mc = mc + req := &heartbeatpb.MaintainerBootstrapRequest{ + ChangefeedID: cfID.ToPB(), + Config: configBytes, + StartTs: 100, + MaintainerEpoch: tc.oldEpoch, + } + require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("old-maintainer"), req)) + requireMessage(t, mc.GetMessageChannel(), "initial bootstrap response was not sent") + manager := orchestrator.dispatcherManagers[cfID] + cleanupDispatcherManager(t, manager) + require.NoError(t, orchestrator.handleCloseRequest(node.ID("old-maintainer"), &heartbeatpb.MaintainerCloseRequest{ + ChangefeedID: cfID.ToPB(), + MaintainerEpoch: tc.oldEpoch, + Removed: tc.removed, + })) + msg := requireMessage(t, mc.GetMessageChannel(), "close response was not sent") + require.False(t, msg.Message[0].(*heartbeatpb.MaintainerCloseResponse).Success) + require.Contains(t, orchestrator.closedMaintainerEpochs, cfID) + require.Equal(t, tc.oldEpoch, orchestrator.closedMaintainerEpochs[cfID]) + require.Eventually(t, func() bool { + return manager.TryClose(false) + }, time.Second, 10*time.Millisecond) + + req.MaintainerEpoch = tc.requestEpoch + require.NoError(t, orchestrator.handleBootstrapRequest(node.ID("old-maintainer"), req)) + require.Empty(t, mc.GetMessageChannel(), "a closed maintainer epoch must not bootstrap again") + if remaining, exists := orchestrator.dispatcherManagers[cfID]; exists { + require.Same(t, manager, remaining) + require.True(t, remaining.IsClosing()) + } + }) + } +} + func TestHandlePostBootstrapWritePathClosedReleasesMaintainerFence(t *testing.T) { mc := messaging.NewMockMessageCenter() setupBootstrapTestServices(t, mc, true) From 54bf027182bde72cfd220b2bd31c2b057b674cc3 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 23 Oct 2026 16:39:21 +0800 Subject: [PATCH 2/2] rename method --- downstreamadapter/dispatchermanager/dispatcher_manager.go | 4 ++-- .../dispatchermanager/dispatcher_manager_test.go | 6 +++--- .../dispatcherorchestrator/dispatcher_orchestrator.go | 2 +- .../dispatcherorchestrator/dispatcher_orchestrator_test.go | 2 +- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/downstreamadapter/dispatchermanager/dispatcher_manager.go b/downstreamadapter/dispatchermanager/dispatcher_manager.go index f801d2c819..78bf5672bf 100644 --- a/downstreamadapter/dispatchermanager/dispatcher_manager.go +++ b/downstreamadapter/dispatchermanager/dispatcher_manager.go @@ -1054,8 +1054,8 @@ func (e *DispatcherManager) mergeEventDispatcher(dispatcherIDs []common.Dispatch // ==== remove and clean related functions ==== -// IsClosing reports whether shutdown has started, including a completed shutdown. -func (e *DispatcherManager) IsClosing() bool { +// IsClosingOrClosed reports whether shutdown has started or completed. +func (e *DispatcherManager) IsClosingOrClosed() bool { return e.closing.Load() || e.closed.Load() } diff --git a/downstreamadapter/dispatchermanager/dispatcher_manager_test.go b/downstreamadapter/dispatchermanager/dispatcher_manager_test.go index c766e6cc36..db9a5a8101 100644 --- a/downstreamadapter/dispatchermanager/dispatcher_manager_test.go +++ b/downstreamadapter/dispatchermanager/dispatcher_manager_test.go @@ -518,7 +518,7 @@ func TestTryCloseRemovedRequestAfterClosedReturnsImmediatelyAndTriggersCleanup(t func TestLocalFenceCancelsWritePathWithoutWaitingForCleanup(t *testing.T) { manager := createTestManager(t) - require.False(t, manager.IsClosing()) + require.False(t, manager.IsClosingOrClosed()) ctx, cancel := context.WithCancel(context.Background()) manager.ctx = ctx manager.cancel = cancel @@ -536,14 +536,14 @@ func TestLocalFenceCancelsWritePathWithoutWaitingForCleanup(t *testing.T) { require.FailNow(t, "local fence should not wait for dispatcher manager cleanup") } require.ErrorIs(t, ctx.Err(), context.Canceled) - require.True(t, manager.IsClosing()) + require.True(t, manager.IsClosingOrClosed()) require.False(t, manager.TryClose(false)) manager.wg.Done() require.Eventually(t, func() bool { return manager.TryClose(false) }, time.Second, 10*time.Millisecond) - require.True(t, manager.IsClosing()) + require.True(t, manager.IsClosingOrClosed()) } func TestLocalFenceDoesNotWaitForBootstrapWriteBlockEvent(t *testing.T) { diff --git a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go index bb1a0715bb..7c3894133f 100644 --- a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go +++ b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go @@ -197,7 +197,7 @@ func (m *DispatcherOrchestrator) handleBootstrapRequest( closedEpoch, closed := m.closedMaintainerEpochs[cfId] m.mutex.Unlock() - if exists && manager.IsClosing() { + if exists && manager.IsClosingOrClosed() { manager.MaintainerFenceMu.Lock() if !manager.CanUpdateMaintainer(from, maintainerEpoch) || !manager.TryClose(false) { manager.MaintainerFenceMu.Unlock() diff --git a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go index 06e47096ae..da90310a81 100644 --- a/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go +++ b/downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator_test.go @@ -978,7 +978,7 @@ func TestHandleBootstrapRequestRejectsStaleEpochAfterManagerClose(t *testing.T) require.Empty(t, mc.GetMessageChannel(), "a closed maintainer epoch must not bootstrap again") if remaining, exists := orchestrator.dispatcherManagers[cfID]; exists { require.Same(t, manager, remaining) - require.True(t, remaining.IsClosing()) + require.True(t, remaining.IsClosingOrClosed()) } }) }