diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go
index 0cf1356254..fa68a1d36e 100644
--- a/cmd/kafka-consumer/writer.go
+++ b/cmd/kafka-consumer/writer.go
@@ -64,10 +64,8 @@ func (p *partitionProgress) updateWatermark(newWatermark uint64, offset kafka.Of
 			zap.Uint64("watermark", newWatermark))
 		return
 	}
-	readOldOffset := true
-	if offset > p.watermarkOffset {
-		readOldOffset = false
-	}
+	readOldOffset := offset <= p.watermarkOffset
+
 	log.Warn("partition resolved ts fall back, ignore it",
 		zap.Bool("readOldOffset", readOldOffset),
 		zap.Int32("partition", p.partition),
@@ -121,7 +119,8 @@ func newWriter(ctx context.Context, o *option) *writer {
 		w.progresses[i] = newPartitionProgress(int32(i), decoder)
 	}
 
-	eventRouter, err := eventrouter.NewEventRouter(o.sinkConfig, o.topic, false, o.protocol == config.ProtocolAvro)
+	isAvroLike := o.protocol == config.ProtocolAvro
+	eventRouter, err := eventrouter.NewEventRouter(o.sinkConfig, o.topic, false, isAvroLike)
 	if err != nil {
 		log.Panic("initialize the event router failed",
 			zap.Any("protocol", o.protocol), zap.Any("topic", o.topic),
@@ -158,12 +157,6 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error {
 	tableIDs := w.getBlockTableIDs(ddl)
 	commitTs := ddl.GetCommitTs()
 	resolvedEvents := make([]*event.DMLEvent, 0)
-	// resolvedGroups records which EventsGroup has flushed events so we can
-	// advance its AppliedWatermark after the flush is fully finished.
-	resolvedGroups := make([]struct {
-		group       *util.EventsGroup
-		maxCommitTs uint64
-	}, 0)
 	for tableID := range tableIDs {
 		for _, progress := range w.progresses {
 			g, ok := progress.eventsGroup[tableID]
@@ -204,11 +197,6 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error {
 			log.Info("flush DML events before DDL done", zap.Uint64("DDLCommitTs", commitTs),
 				zap.Int("total", total), zap.Duration("duration", time.Since(start)),
 				zap.Any("tables", tableIDs))
-			for _, item := range resolvedGroups {
-				if item.maxCommitTs > item.group.AppliedWatermark {
-					item.group.AppliedWatermark = item.maxCommitTs
-				}
-			}
 			return w.mysqlSink.WriteBlockEvent(ddl)
 		case <-ticker.C:
 			log.Warn("DML events cannot be flushed in time",
@@ -285,12 +273,6 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error {
 
 	watermark := w.globalWatermark()
 	resolvedEvents := make([]*event.DMLEvent, 0)
-	// resolvedGroups records which EventsGroup has flushed events so we can
-	// advance its AppliedWatermark after the flush is fully finished.
-	resolvedGroups := make([]struct {
-		group       *util.EventsGroup
-		maxCommitTs uint64
-	}, 0)
 	for _, p := range w.progresses {
 		for _, group := range p.eventsGroup {
 			messages := group.ResolveInto(watermark, nil)
@@ -327,11 +309,6 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error {
 		case <-done:
 			log.Info("flush DML events done", zap.Uint64("watermark", watermark),
 				zap.Int("total", total), zap.Duration("duration", time.Since(start)))
-			for _, item := range resolvedGroups {
-				if item.maxCommitTs > item.group.AppliedWatermark {
-					item.group.AppliedWatermark = item.maxCommitTs
-				}
-			}
 			return nil
 		case <-ticker.C:
 			log.Warn("DML events cannot be flushed in time", zap.Uint64("watermark", watermark),
@@ -632,84 +609,53 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti
 		table    = message.Table
 		commitTs = message.GetCommitTs()
 	)
+	globalWatermark := w.globalWatermark()
+	if commitTs < globalWatermark {
+		log.Warn("DML event fallback row, since less than the global watermark, ignore it",
+			zap.Int64("tableID", tableID), zap.Int32("partition", progress.partition),
+			zap.Uint64("commitTs", commitTs), zap.Any("offset", offset),
+			zap.Uint64("globalWatermark", globalWatermark),
+			zap.Uint64("partitionWatermark", progress.watermark),
+			zap.Any("watermarkOffset", progress.watermarkOffset),
+			zap.String("schema", schema), zap.String("table", table),
+			zap.Stringer("eventType", message.RowType),
+			zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes))
+		return
+	}
+
 	group := progress.eventsGroup[tableID]
 	if group == nil {
 		group = util.NewEventsGroup(progress.partition, tableID)
 		progress.eventsGroup[tableID] = group
 	}
-	// IMPORTANT: Kafka offsets are append-only, but CommitTs can go backwards after
-	// a TiCDC restart/retry (at-least-once replay). We must not drop such events
-	// solely based on a "seen" watermark (e.g. HighWatermark). The only safe
-	// ignore condition is "already flushed to downstream".
-	if commitTs <= group.AppliedWatermark {
-		log.Warn("DML event replayed after applied, ignore it",
+	message = w.messageWithPartitionCheck(message, progress.partition, offset)
+	group.AppendMessage(message)
+	if commitTs < progress.watermark {
+		log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush",
 			zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition),
 			zap.Uint64("commitTs", commitTs), zap.Any("offset", offset),
-			zap.Uint64("appliedWatermark", group.AppliedWatermark), zap.Uint64("highWatermark", group.HighWatermark),
-			zap.Uint64("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
-			zap.String("schema", schema), zap.String("table", table), zap.Any("protocol", w.protocol))
+			zap.Uint64("watermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
+			zap.Uint64("globalWatermark", globalWatermark),
+			zap.String("schema", schema), zap.String("table", table),
+			zap.Stringer("eventType", message.RowType),
+			zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes))
 		return
 	}
 	if commitTs >= group.HighWatermark {
-		message = w.messageWithPartitionCheck(message, progress.partition, offset)
-		group.AppendMessage(message, false)
 		log.Debug("DML event append to the group",
-			zap.Int32("partition", group.Partition), zap.Any("offset", offset),
-			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
-			zap.Uint64("appliedWatermark", group.AppliedWatermark),
-			zap.Uint64("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
-			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-			zap.Stringer("eventType", message.RowType))
-		return
-	}
-	if w.enableTableAcrossNodes {
-		log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it",
 			zap.Int32("partition", group.Partition), zap.Any("offset", offset),
 			zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark),
 			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
 			zap.Stringer("eventType", message.RowType))
-		group.AppendMessage(w.messageWithPartitionCheck(message, progress.partition, offset), true)
 		return
 	}
-	switch w.protocol {
-	case config.ProtocolSimple:
-		// simple protocol set the table id for all row message, it can be known which table the row message belongs to,
-		// also consider the table partition.
-		// open protocol set the partition table id if the table is partitioned.
-		// for normal table, the table id is generated by the fake table id generator by using schema and table name.
-		// so one event group for one normal table or one table partition, replayed messages can be ignored.
-		log.Warn("DML event fallback row, since less than the group high watermark, ignore it",
-			zap.Int32("partition", progress.partition), zap.Any("offset", offset),
-			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
-			zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
-			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-			zap.Stringer("eventType", message.RowType),
-			// zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns),
-			zap.Any("protocol", w.protocol))
-	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro,
-		config.ProtocolDebezium:
-		// for partition table, these protocols cannot assign physical table id to each dml message,
-		// we cannot distinguish whether it's a real fallback event or not, still append it.
-		if w.partitionTableAccessor.IsPartitionTable(schema, table) {
-			log.Warn("DML events fallback, but the table is a partition table, still append it",
-				zap.Int32("partition", group.Partition), zap.Any("offset", offset),
-				zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
-				zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-				zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol))
-			group.AppendMessage(w.messageWithPartitionCheck(message, progress.partition, offset), true)
-			return
-		}
-		log.Warn("DML event fallback row, since less than the group high watermark, ignore it",
-			zap.Int32("partition", progress.partition), zap.Any("offset", offset),
-			zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark),
-			zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
-			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-			zap.Stringer("eventType", message.RowType),
-			// zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns),
-			zap.Any("protocol", w.protocol))
-	default:
-		log.Panic("unknown protocol", zap.Any("protocol", w.protocol))
-	}
+	log.Warn("DML event commit ts fallback, append it and sort before flush",
+		zap.Int32("partition", progress.partition), zap.Any("offset", offset),
+		zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
+		zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
+		zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
+		zap.Stringer("eventType", message.RowType),
+		zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes))
 }
 
 func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go
index 616b450b26..e5845efb03 100644
--- a/cmd/kafka-consumer/writer_test.go
+++ b/cmd/kafka-consumer/writer_test.go
@@ -18,9 +18,10 @@ import (
 	"testing"
 
 	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"github.com/golang/mock/gomock"
 	"github.com/pingcap/ticdc/cmd/util"
-	"github.com/pingcap/ticdc/downstreamadapter/sink"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
+	sinkmock "github.com/pingcap/ticdc/downstreamadapter/sink/mock"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
@@ -30,40 +31,23 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// recordingSink is a minimal sink.Sink implementation that records which DDLs are executed.
-//
-// It lets unit tests validate consumer-side DDL flushing behavior without requiring a real downstream.
-type recordingSink struct {
-	ddls []string
-}
-
-var _ sink.Sink = (*recordingSink)(nil)
-
-func (s *recordingSink) SinkType() common.SinkType { return common.MysqlSinkType }
-func (s *recordingSink) IsNormal() bool            { return true }
-func (s *recordingSink) AddDMLEvent(_ *commonEvent.DMLEvent) {
-}
-
-func (s *recordingSink) FlushDMLBeforeBlock(_ commonEvent.BlockEvent) error {
-	return nil
-}
-
-func (s *recordingSink) WriteBlockEvent(event commonEvent.BlockEvent) error {
-	if ddl, ok := event.(*commonEvent.DDLEvent); ok {
-		s.ddls = append(s.ddls, ddl.Query)
-	}
-	return nil
-}
+func newMockSink(t *testing.T) (*sinkmock.MockSink, *[]string) {
+	t.Helper()
 
-func (s *recordingSink) AddCheckpointTs(_ uint64) {
-}
+	ctrl := gomock.NewController(t)
+	s := sinkmock.NewMockSink(ctrl)
+	ddls := make([]string, 0)
 
-func (s *recordingSink) SetTableSchemaStore(_ *commonEvent.TableSchemaStore) {
-}
+	s.EXPECT().AddDMLEvent(gomock.Any()).AnyTimes()
+	s.EXPECT().WriteBlockEvent(gomock.Any()).DoAndReturn(func(event commonEvent.BlockEvent) error {
+		if ddl, ok := event.(*commonEvent.DDLEvent); ok {
+			ddls = append(ddls, ddl.Query)
+		}
+		return nil
+	}).AnyTimes()
 
-func (s *recordingSink) Close() {
+	return s, &ddls
 }
-func (s *recordingSink) Run(_ context.Context) error { return nil }
 
 func TestWriterWrite_executesIndependentCreateTableWithoutWatermark(t *testing.T) {
 	// Scenario: In some integration tests the upstream intentionally pauses dispatcher creation, which can
@@ -75,7 +59,7 @@ func TestWriterWrite_executesIndependentCreateTableWithoutWatermark(t *testing.T
 	// 2) Call writer.Write and expect the DDL is executed to advance downstream schema even without the
 	//    watermark catching up.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	w := &writer{
 		progresses: []*partitionProgress{
 			{partition: 0, watermark: 0},
@@ -100,7 +84,7 @@ func TestWriterWrite_executesIndependentCreateTableWithoutWatermark(t *testing.T
 
 	w.Write(ctx, codeccommon.MessageTypeDDL)
 
-	require.Equal(t, []string{"CREATE TABLE `test`.`t` (`id` INT PRIMARY KEY)"}, s.ddls)
+	require.Equal(t, []string{"CREATE TABLE `test`.`t` (`id` INT PRIMARY KEY)"}, *ddls)
 	require.Empty(t, w.ddlList)
 }
 
@@ -113,7 +97,7 @@ func TestWriterWrite_preservesOrderWhenBlockedDDLNotReady(t *testing.T) {
 	// 2) Call writer.Write and expect nothing executes.
 	// 3) Advance watermark beyond the first DDL and expect both execute in order.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	p := &partitionProgress{partition: 0, watermark: 0}
 	w := &writer{
 		progresses: []*partitionProgress{p},
@@ -145,7 +129,7 @@ func TestWriterWrite_preservesOrderWhenBlockedDDLNotReady(t *testing.T) {
 	}
 
 	w.Write(ctx, codeccommon.MessageTypeDDL)
-	require.Empty(t, s.ddls)
+	require.Empty(t, *ddls)
 	require.Len(t, w.ddlList, 2)
 
 	p.watermark = 200
@@ -153,7 +137,7 @@ func TestWriterWrite_preservesOrderWhenBlockedDDLNotReady(t *testing.T) {
 	require.Equal(t, []string{
 		"ALTER TABLE `test`.`t` ADD COLUMN `c2` INT",
 		"CREATE TABLE `test`.`t2` (`id` INT PRIMARY KEY)",
-	}, s.ddls)
+	}, *ddls)
 	require.Empty(t, w.ddlList)
 }
 
@@ -166,7 +150,7 @@ func TestWriterWrite_doesNotBypassWatermarkForCreateTableLike(t *testing.T) {
 	// 2) Call writer.Write and expect the DDL is NOT executed.
 	// 3) Advance watermark beyond the DDL commitTs and expect the DDL executes.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	p := &partitionProgress{partition: 0, watermark: 0}
 	w := &writer{
 		progresses: []*partitionProgress{p},
@@ -191,12 +175,12 @@ func TestWriterWrite_doesNotBypassWatermarkForCreateTableLike(t *testing.T) {
 	}
 
 	w.Write(ctx, codeccommon.MessageTypeDDL)
-	require.Empty(t, s.ddls)
+	require.Empty(t, *ddls)
 	require.Len(t, w.ddlList, 1)
 
 	p.watermark = 200
 	w.Write(ctx, codeccommon.MessageTypeDDL)
-	require.Equal(t, []string{"CREATE TABLE `test`.`t2` LIKE `test`.`t1`"}, s.ddls)
+	require.Equal(t, []string{"CREATE TABLE `test`.`t2` LIKE `test`.`t1`"}, *ddls)
 	require.Empty(t, w.ddlList)
 }
 
@@ -210,7 +194,7 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) {
 	// 2) Call writer.Write and expect all DDLs with commitTs <= watermark execute (in commit-ts order),
 	//    and only the truly "future" DDL remains pending.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	p := &partitionProgress{partition: 0, watermark: 944040962}
 	w := &writer{
 		progresses: []*partitionProgress{p},
@@ -279,11 +263,101 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) {
 		"ALTER TABLE `common_1`.`add_and_drop_columns` ADD COLUMN `col1` INT NULL, ADD COLUMN `col2` INT NULL, ADD COLUMN `col3` INT NULL",
 		"ALTER TABLE `common_1`.`add_and_drop_columns` DROP COLUMN `col1`, DROP COLUMN `col2`",
 		"CREATE DATABASE `common`",
-	}, s.ddls)
+	}, *ddls)
 	require.Len(t, w.ddlList, 1)
 	require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query)
 }
 
+func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	s := sinkmock.NewMockSink(ctrl)
+	flushedCommitTs := make([]uint64, 0)
+	s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) {
+		flushedCommitTs = append(flushedCommitTs, event.GetCommitTs())
+		event.PostFlush()
+	}).Times(2)
+
+	replicaCfg := config.GetDefaultReplicaConfig()
+	eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false)
+	require.NoError(t, err)
+
+	p := &partitionProgress{
+		partition:   0,
+		eventsGroup: make(map[int64]*util.EventsGroup),
+		watermark:   0,
+	}
+	w := &writer{
+		progresses:  []*partitionProgress{p},
+		mysqlSink:   s,
+		eventRouter: eventRouter,
+		protocol:    config.ProtocolOpen,
+	}
+
+	w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(1))
+	w.appendMessage2Group(newDMLMessageForWriterTest(10), p, kafka.Offset(2))
+	w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(3))
+
+	p.watermark = 20
+	require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved))
+	require.Equal(t, []uint64{10, 20}, flushedCommitTs)
+}
+
+func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	s := sinkmock.NewMockSink(ctrl)
+	s.EXPECT().AddDMLEvent(gomock.Any()).Times(0)
+
+	progress := &partitionProgress{
+		partition:   0,
+		eventsGroup: make(map[int64]*util.EventsGroup),
+		watermark:   20,
+		decoder:     &singleDMLDecoder{message: newDMLMessageForWriterTest(10)},
+	}
+	w := &writer{
+		progresses:      []*partitionProgress{progress},
+		mysqlSink:       s,
+		protocol:        config.ProtocolOpen,
+		maxBatchSize:    64,
+		maxMessageBytes: 1,
+	}
+
+	needCommit := w.WriteMessage(ctx, &kafka.Message{
+		TopicPartition: kafka.TopicPartition{Partition: 0, Offset: kafka.Offset(10)},
+	})
+
+	require.False(t, needCommit)
+	require.Nil(t, progress.eventsGroup[1])
+}
+
+func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) {
+	replicaCfg := config.GetDefaultReplicaConfig()
+	eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false)
+	require.NoError(t, err)
+
+	progress := &partitionProgress{
+		partition:   0,
+		eventsGroup: make(map[int64]*util.EventsGroup),
+		watermark:   20,
+	}
+	w := &writer{
+		progresses: []*partitionProgress{
+			progress,
+			{partition: 1, watermark: 5},
+		},
+		eventRouter: eventRouter,
+		protocol:    config.ProtocolOpen,
+	}
+
+	w.appendMessage2Group(newDMLMessageForWriterTest(10), progress, kafka.Offset(10))
+
+	require.NotNil(t, progress.eventsGroup[1])
+	resolved := progress.eventsGroup[1].ResolveInto(20, nil)
+	require.Len(t, resolved, 1)
+	require.Equal(t, uint64(10), resolved[0].GetCommitTs())
+}
+
 func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) {
 	replicaCfg := config.GetDefaultReplicaConfig()
 	eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, true)
@@ -383,3 +457,42 @@ func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) {
 		})
 	}
 }
+
+func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage {
+	return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent {
+		return &commonEvent.DMLEvent{
+			PhysicalTableID: 1,
+			CommitTs:        commitTs,
+			RowTypes:        []common.RowType{common.RowTypeUpdate},
+			Rows:            chunk.NewChunkWithCapacity(nil, 0),
+			TableInfo: &common.TableInfo{
+				TableName: common.TableName{Schema: "test", Table: "t", TableID: 1},
+			},
+		}
+	})
+}
+
+type singleDMLDecoder struct {
+	message  *codeccommon.DMLMessage
+	consumed bool
+}
+
+func (d *singleDMLDecoder) AddKeyValue(_, _ []byte) {
+}
+
+func (d *singleDMLDecoder) HasNext() (codeccommon.MessageType, bool) {
+	return codeccommon.MessageTypeRow, !d.consumed
+}
+
+func (d *singleDMLDecoder) NextResolvedEvent() uint64 {
+	return 0
+}
+
+func (d *singleDMLDecoder) NextDMLMessage() *codeccommon.DMLMessage {
+	d.consumed = true
+	return d.message
+}
+
+func (d *singleDMLDecoder) NextDDLEvent() *commonEvent.DDLEvent {
+	return nil
+}
diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go
index 1210dd62a8..afd0aeddca 100644
--- a/cmd/pulsar-consumer/writer.go
+++ b/cmd/pulsar-consumer/writer.go
@@ -149,12 +149,6 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) e
 	tableIDs := w.getBlockTableIDs(ddl)
 	commitTs := ddl.GetCommitTs()
 	resolvedEvents := make([]*commonEvent.DMLEvent, 0)
-	// resolvedGroups records which EventsGroup has flushed events so we can
-	// advance its AppliedWatermark after the flush is fully finished.
-	resolvedGroups := make([]struct {
-		group       *util.EventsGroup
-		maxCommitTs uint64
-	}, 0)
 	for tableID := range tableIDs {
 		for _, progress := range w.progresses {
 			g, ok := progress.eventsGroup[tableID]
@@ -195,11 +189,6 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) e
 			log.Info("flush DML events before DDL done", zap.Uint64("DDLCommitTs", commitTs),
 				zap.Int("total", total), zap.Duration("duration", time.Since(start)),
 				zap.Any("tables", tableIDs))
-			for _, item := range resolvedGroups {
-				if item.maxCommitTs > item.group.AppliedWatermark {
-					item.group.AppliedWatermark = item.maxCommitTs
-				}
-			}
 			return w.mysqlSink.WriteBlockEvent(ddl)
 		case <-ticker.C:
 			log.Warn("DML events cannot be flushed in time",
@@ -276,12 +265,6 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error {
 
 	watermark := w.globalWatermark()
 	resolvedEvents := make([]*commonEvent.DMLEvent, 0)
-	// resolvedGroups records which EventsGroup has flushed events so we can
-	// advance its AppliedWatermark after the flush is fully finished.
-	resolvedGroups := make([]struct {
-		group       *util.EventsGroup
-		maxCommitTs uint64
-	}, 0)
 	for _, p := range w.progresses {
 		for _, group := range p.eventsGroup {
 			messages := group.ResolveInto(watermark, nil)
@@ -316,11 +299,6 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error {
 		case <-done:
 			log.Info("flush DML events done", zap.Uint64("watermark", watermark),
 				zap.Int("total", total), zap.Duration("duration", time.Since(start)))
-			for _, item := range resolvedGroups {
-				if item.maxCommitTs > item.group.AppliedWatermark {
-					item.group.AppliedWatermark = item.maxCommitTs
-				}
-			}
 			return nil
 		case <-ticker.C:
 			log.Warn("DML events cannot be flushed in time", zap.Uint64("watermark", watermark),
@@ -520,59 +498,47 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti
 		table    = message.Table
 		commitTs = message.GetCommitTs()
 	)
+	globalWatermark := w.globalWatermark()
+	if commitTs < globalWatermark {
+		log.Warn("DML event fallback row, since less than the global watermark, ignore it",
+			zap.Int64("tableID", tableID), zap.Int32("partition", progress.partition),
+			zap.Uint64("commitTs", commitTs),
+			zap.Uint64("globalWatermark", globalWatermark),
+			zap.Uint64("partitionWatermark", progress.watermark),
+			zap.String("schema", schema), zap.String("table", table),
+			zap.Stringer("eventType", message.RowType),
+			zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes))
+		return
+	}
+
 	group := progress.eventsGroup[tableID]
 	if group == nil {
 		group = util.NewEventsGroup(progress.partition, tableID)
 		progress.eventsGroup[tableID] = group
 	}
-	if commitTs <= group.AppliedWatermark {
-		log.Warn("DML event replayed after applied, ignore it",
+	group.AppendMessage(message)
+	if commitTs < progress.watermark {
+		log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush",
 			zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition),
-			zap.Uint64("commitTs", commitTs),
-			zap.Uint64("appliedWatermark", group.AppliedWatermark), zap.Uint64("highWatermark", group.HighWatermark),
-			zap.Uint64("partitionWatermark", progress.watermark),
-			zap.String("schema", schema), zap.String("table", table), zap.Any("protocol", w.protocol))
+			zap.Uint64("commitTs", commitTs), zap.Uint64("watermark", progress.watermark),
+			zap.Uint64("globalWatermark", globalWatermark),
+			zap.String("schema", schema), zap.String("table", table),
+			zap.Stringer("eventType", message.RowType),
+			zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes))
 		return
 	}
 	if commitTs >= group.HighWatermark {
-		group.AppendMessage(message, false)
 		log.Debug("DML event append to the group",
-			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
-			zap.Uint64("appliedWatermark", group.AppliedWatermark),
-			zap.Uint64("partitionWatermark", progress.watermark),
-			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-			zap.Stringer("eventType", message.RowType))
-		return
-	}
-	if w.enableTableAcrossNodes {
-		log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it",
 			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
 			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
 			zap.Stringer("eventType", message.RowType))
-		group.AppendMessage(message, true)
 		return
 	}
-	switch w.protocol {
-	case config.ProtocolCanalJSON:
-		// for partition table, the canal-json message cannot assign physical table id to each dml message,
-		// we cannot distinguish whether it's a real fallback event or not, still append it.
-		isPartitionTable := w.partitionTableAccessor != nil &&
-			w.partitionTableAccessor.IsPartitionTable(schema, table)
-		if isPartitionTable {
-			log.Warn("DML events fallback, but it's canal-json and partition table, still append it",
-				zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
-				zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-				zap.Stringer("eventType", message.RowType))
-			group.AppendMessage(message, true)
-			return
-		}
-		log.Warn("DML event fallback row, since less than the group high watermark, ignore it",
-			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
-			zap.Any("partitionWatermark", progress.watermark), zap.Any("watermark", progress.watermark),
-			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
-			zap.Stringer("eventType", message.RowType),
-			zap.Any("protocol", w.protocol), zap.Bool("IsPartition", isPartitionTable))
-	default:
-		log.Panic("unknown protocol", zap.Any("protocol", w.protocol))
-	}
+	log.Warn("DML event commit ts fallback, append it and sort before flush",
+		zap.Int32("partition", progress.partition),
+		zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
+		zap.Any("partitionWatermark", progress.watermark),
+		zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
+		zap.Stringer("eventType", message.RowType),
+		zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes))
 }
diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go
index 39a302c8e1..35c9c037e3 100644
--- a/cmd/pulsar-consumer/writer_test.go
+++ b/cmd/pulsar-consumer/writer_test.go
@@ -21,50 +21,33 @@ import (
 	"github.com/apache/pulsar-client-go/pulsar"
 	"github.com/golang/mock/gomock"
 	"github.com/pingcap/ticdc/cmd/util"
-	"github.com/pingcap/ticdc/downstreamadapter/sink"
 	sinkmock "github.com/pingcap/ticdc/downstreamadapter/sink/mock"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	codeccommon "github.com/pingcap/ticdc/pkg/sink/codec/common"
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/util/chunk"
 	"github.com/stretchr/testify/require"
 )
 
-// recordingSink is a minimal sink.Sink implementation that records which DDLs are executed.
-//
-// It lets unit tests validate consumer-side DDL flushing behavior without requiring a real downstream.
-type recordingSink struct {
-	ddls []string
-}
-
-var _ sink.Sink = (*recordingSink)(nil)
-
-func (s *recordingSink) SinkType() common.SinkType { return common.MysqlSinkType }
-func (s *recordingSink) IsNormal() bool            { return true }
-func (s *recordingSink) AddDMLEvent(_ *commonEvent.DMLEvent) {
-}
-
-func (s *recordingSink) FlushDMLBeforeBlock(_ commonEvent.BlockEvent) error {
-	return nil
-}
-
-func (s *recordingSink) WriteBlockEvent(event commonEvent.BlockEvent) error {
-	if ddl, ok := event.(*commonEvent.DDLEvent); ok {
-		s.ddls = append(s.ddls, ddl.Query)
-	}
-	return nil
-}
+func newMockSink(t *testing.T) (*sinkmock.MockSink, *[]string) {
+	t.Helper()
 
-func (s *recordingSink) AddCheckpointTs(_ uint64) {
-}
+	ctrl := gomock.NewController(t)
+	s := sinkmock.NewMockSink(ctrl)
+	ddls := make([]string, 0)
 
-func (s *recordingSink) SetTableSchemaStore(_ *commonEvent.TableSchemaStore) {
-}
+	s.EXPECT().AddDMLEvent(gomock.Any()).AnyTimes()
+	s.EXPECT().WriteBlockEvent(gomock.Any()).DoAndReturn(func(event commonEvent.BlockEvent) error {
+		if ddl, ok := event.(*commonEvent.DDLEvent); ok {
+			ddls = append(ddls, ddl.Query)
+		}
+		return nil
+	}).AnyTimes()
 
-func (s *recordingSink) Close() {
+	return s, &ddls
 }
-func (s *recordingSink) Run(_ context.Context) error { return nil }
 
 func TestWriterWrite_executesIndependentCreateTableWithoutWatermark(t *testing.T) {
 	// Scenario: If upstream resolved-ts is held back (e.g. failpoints in integration tests), the consumer
@@ -75,7 +58,7 @@ func TestWriterWrite_executesIndependentCreateTableWithoutWatermark(t *testing.T
 	// 1) Enqueue an independent CREATE TABLE DDL with commitTs > watermark.
 	// 2) Call writer.Write and expect the DDL is executed even without watermark catching up.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	w := &writer{
 		progresses: []*partitionProgress{
 			{partition: 0, watermark: 0},
@@ -100,7 +83,7 @@ func TestWriterWrite_executesIndependentCreateTableWithoutWatermark(t *testing.T
 
 	w.Write(ctx, codeccommon.MessageTypeDDL)
 
-	require.Equal(t, []string{"CREATE TABLE `test`.`t` (`id` INT PRIMARY KEY)"}, s.ddls)
+	require.Equal(t, []string{"CREATE TABLE `test`.`t` (`id` INT PRIMARY KEY)"}, *ddls)
 	require.Empty(t, w.ddlList)
 }
 
@@ -113,7 +96,7 @@ func TestWriterWrite_preservesOrderWhenBlockedDDLNotReady(t *testing.T) {
 	// 2) Call writer.Write and expect nothing executes.
 	// 3) Advance watermark beyond the first DDL and expect both execute in order.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	p := &partitionProgress{partition: 0, watermark: 0}
 	w := &writer{
 		progresses: []*partitionProgress{p},
@@ -145,7 +128,7 @@ func TestWriterWrite_preservesOrderWhenBlockedDDLNotReady(t *testing.T) {
 	}
 
 	w.Write(ctx, codeccommon.MessageTypeDDL)
-	require.Empty(t, s.ddls)
+	require.Empty(t, *ddls)
 	require.Len(t, w.ddlList, 2)
 
 	p.watermark = 200
@@ -153,7 +136,7 @@ func TestWriterWrite_preservesOrderWhenBlockedDDLNotReady(t *testing.T) {
 	require.Equal(t, []string{
 		"ALTER TABLE `test`.`t` ADD COLUMN `c2` INT",
 		"CREATE TABLE `test`.`t2` (`id` INT PRIMARY KEY)",
-	}, s.ddls)
+	}, *ddls)
 	require.Empty(t, w.ddlList)
 }
 
@@ -166,7 +149,7 @@ func TestWriterWrite_doesNotBypassWatermarkForCreateTableLike(t *testing.T) {
 	// 2) Call writer.Write and expect the DDL is NOT executed.
 	// 3) Advance watermark beyond the DDL commitTs and expect the DDL executes.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	p := &partitionProgress{partition: 0, watermark: 0}
 	w := &writer{
 		progresses: []*partitionProgress{p},
@@ -191,12 +174,12 @@ func TestWriterWrite_doesNotBypassWatermarkForCreateTableLike(t *testing.T) {
 	}
 
 	w.Write(ctx, codeccommon.MessageTypeDDL)
-	require.Empty(t, s.ddls)
+	require.Empty(t, *ddls)
 	require.Len(t, w.ddlList, 1)
 
 	p.watermark = 200
 	w.Write(ctx, codeccommon.MessageTypeDDL)
-	require.Equal(t, []string{"CREATE TABLE `test`.`t2` LIKE `test`.`t1`"}, s.ddls)
+	require.Equal(t, []string{"CREATE TABLE `test`.`t2` LIKE `test`.`t1`"}, *ddls)
 	require.Empty(t, w.ddlList)
 }
 
@@ -211,7 +194,7 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) {
 	// 2) Call writer.Write and expect all DDLs with commitTs <= watermark execute (in commit-ts order),
 	//    and only the truly "future" DDL remains pending.
 	ctx := context.Background()
-	s := &recordingSink{}
+	s, ddls := newMockSink(t)
 	p := &partitionProgress{partition: 0, watermark: 944040962}
 	w := &writer{
 		progresses: []*partitionProgress{p},
@@ -280,18 +263,101 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) {
 		"ALTER TABLE `common_1`.`add_and_drop_columns` ADD COLUMN `col1` INT NULL, ADD COLUMN `col2` INT NULL, ADD COLUMN `col3` INT NULL",
 		"ALTER TABLE `common_1`.`add_and_drop_columns` DROP COLUMN `col1`, DROP COLUMN `col2`",
 		"CREATE DATABASE `common`",
-	}, s.ddls)
+	}, *ddls)
 	require.Len(t, w.ddlList, 1)
 	require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query)
 }
 
+func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	s := sinkmock.NewMockSink(ctrl)
+	flushedCommitTs := make([]uint64, 0)
+	s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) {
+		flushedCommitTs = append(flushedCommitTs, event.GetCommitTs())
+		event.PostFlush()
+	}).Times(2)
+
+	p := &partitionProgress{
+		partition:   0,
+		eventsGroup: make(map[int64]*util.EventsGroup),
+		watermark:   0,
+	}
+	w := &writer{
+		progresses: []*partitionProgress{p},
+		mysqlSink:  s,
+		protocol:   config.ProtocolCanalJSON,
+	}
+
+	w.appendMessage2Group(newDMLMessageForWriterTest(20), p)
+	w.appendMessage2Group(newDMLMessageForWriterTest(10), p)
+	w.appendMessage2Group(newDMLMessageForWriterTest(20), p)
+
+	p.watermark = 20
+	require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved))
+	require.Equal(t, []uint64{10, 20}, flushedCommitTs)
+}
+
+func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	s := sinkmock.NewMockSink(ctrl)
+	s.EXPECT().AddDMLEvent(gomock.Any()).Times(0)
+
+	decoder := &deferredDMLDecoder{
+		row: &commonEvent.DMLEvent{
+			PhysicalTableID: 1,
+			CommitTs:        10,
+			RowTypes:        []common.RowType{common.RowTypeInsert},
+			TableInfo: &common.TableInfo{
+				TableName: common.TableName{Schema: "test", Table: "t", TableID: 1},
+			},
+		},
+	}
+	progress := &partitionProgress{
+		partition:   0,
+		eventsGroup: make(map[int64]*util.EventsGroup),
+		watermark:   20,
+		decoder:     decoder,
+	}
+	w := &writer{
+		progresses: []*partitionProgress{progress},
+		mysqlSink:  s,
+		protocol:   config.ProtocolCanalJSON,
+	}
+
+	needCommit := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)})
+
+	require.False(t, needCommit)
+	require.Nil(t, progress.eventsGroup[1])
+}
+
+func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) {
+	progress := &partitionProgress{
+		partition:   0,
+		eventsGroup: make(map[int64]*util.EventsGroup),
+		watermark:   20,
+	}
+	w := &writer{
+		progresses: []*partitionProgress{
+			progress,
+			{partition: 1, watermark: 5},
+		},
+		protocol: config.ProtocolCanalJSON,
+	}
+
+	w.appendMessage2Group(newDMLMessageForWriterTest(10), progress)
+
+	require.NotNil(t, progress.eventsGroup[1])
+	resolved := progress.eventsGroup[1].ResolveInto(20, nil)
+	require.Len(t, resolved, 1)
+	require.Equal(t, uint64(10), resolved[0].GetCommitTs())
+}
+
 func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) {
 	w := &writer{
 		progresses: []*partitionProgress{
-			{
-				partition:   0,
-				eventsGroup: make(map[int64]*util.EventsGroup),
-			},
+			{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)},
 		},
 		protocol:               config.ProtocolCanalJSON,
 		partitionTableAccessor: codeccommon.NewPartitionTableAccessor(),
@@ -401,7 +467,7 @@ func (d *deferredDMLDecoder) NextResolvedEvent() uint64 {
 
 func (d *deferredDMLDecoder) NextDMLMessage() *codeccommon.DMLMessage {
 	d.nextDMLMessageCount++
-	return codeccommon.NewDMLMessage(1, "test", "t", 100, common.RowTypeInsert, func() *commonEvent.DMLEvent {
+	return codeccommon.NewDMLMessage(1, "test", "t", d.row.CommitTs, common.RowTypeInsert, func() *commonEvent.DMLEvent {
 		d.toDMLEventCount++
 		return d.row
 	})
@@ -411,6 +477,20 @@ func (d *deferredDMLDecoder) NextDDLEvent() *commonEvent.DDLEvent {
 	return nil
 }
 
+func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage {
+	return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent {
+		return &commonEvent.DMLEvent{
+			PhysicalTableID: 1,
+			CommitTs:        commitTs,
+			RowTypes:        []common.RowType{common.RowTypeUpdate},
+			Rows:            chunk.NewChunkWithCapacity(nil, 0),
+			TableInfo: &common.TableInfo{
+				TableName: common.TableName{Schema: "test", Table: "t", TableID: 1},
+			},
+		}
+	})
+}
+
 type fakePulsarMessage struct {
 	key     string
 	payload []byte
diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go
index 03c9e47721..05ec67cf4b 100644
--- a/cmd/storage-consumer/consumer.go
+++ b/cmd/storage-consumer/consumer.go
@@ -250,7 +250,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc
 		c.eventsGroup[tableID] = group
 	}
 	if commitTs >= group.HighWatermark {
-		group.AppendMessage(message, false)
+		group.AppendMessage(message)
 		log.Debug("DML event append to the group",
 			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
 			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
@@ -262,7 +262,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc
 			zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
 			zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
 			zap.Stringer("eventType", message.RowType))
-		group.AppendMessage(message, true)
+		group.AppendMessage(message)
 		return
 	}
 	log.Warn("dml event commit ts fallback, ignore",
diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go
index 31b3c740d5..2391215f03 100644
--- a/cmd/util/event_group.go
+++ b/cmd/util/event_group.go
@@ -14,6 +14,7 @@
 package util
 
 import (
+	"math"
 	"sort"
 
 	"github.com/pingcap/log"
@@ -29,13 +30,6 @@ type EventsGroup struct {
 
 	messages      []*codeccommon.DMLMessage
 	HighWatermark uint64
-	// AppliedWatermark is the maximum CommitTs that has been successfully flushed
-	// to the downstream for this group.
-	//
-	// It is used to distinguish "safe to ignore" replays (CommitTs <=
-	// AppliedWatermark) from "still needed" events that arrive late due to sink
-	// retries / restarts.
-	AppliedWatermark uint64
 }
 
 // NewEventsGroup will create new event group.
@@ -48,65 +42,79 @@ func NewEventsGroup(partition int32, tableID int64) *EventsGroup {
 }
 
 // AppendMessage appends a message to event groups.
-func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage, force bool) {
+func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) {
 	commitTs := message.GetCommitTs()
 	if commitTs > g.HighWatermark {
 		g.HighWatermark = commitTs
 	}
+	g.messages = append(g.messages, message)
+}
 
-	var lastMessage *codeccommon.DMLMessage
-	if len(g.messages) > 0 {
-		lastMessage = g.messages[len(g.messages)-1]
+// ResolveInto appends all messages with CommitTs <= resolve into dst in commit-ts order and removes
+// them from the group. ResolveInto copies pointers into dst first, then clears the resolved messages
+// so Go GC can reclaim them once downstream is done with them.
+func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage {
+	if len(g.messages) == 0 {
+		return dst
 	}
 
-	if lastMessage == nil || lastMessage.GetCommitTs() <= commitTs {
-		g.messages = append(g.messages, message)
-		return
+	original := g.messages
+	remaining := g.messages[:0]
+	resolved := make([]*codeccommon.DMLMessage, 0, len(g.messages))
+
+	var (
+		lastCommitTs       uint64
+		outOfOrder         bool
+		outOfOrderLastTs   uint64
+		outOfOrderCommitTs uint64
+	)
+	for _, message := range g.messages {
+		commitTs := message.GetCommitTs()
+		if commitTs > resolve {
+			remaining = append(remaining, message)
+			continue
+		}
+		if len(resolved) > 0 && commitTs < lastCommitTs && !outOfOrder {
+			outOfOrder = true
+			outOfOrderLastTs = lastCommitTs
+			outOfOrderCommitTs = commitTs
+		}
+		lastCommitTs = commitTs
+		resolved = append(resolved, message)
 	}
-
-	if force {
-		i := sort.Search(len(g.messages), func(i int) bool {
-			return g.messages[i].GetCommitTs() > commitTs
-		})
-		g.messages = append(g.messages, nil)
-		copy(g.messages[i+1:], g.messages[i:])
-		g.messages[i] = message
-		return
+	if len(resolved) == 0 {
+		return dst
 	}
-	log.Panic("append event with smaller commit ts",
-		zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID),
-		zap.Uint64("lastCommitTs", lastMessage.GetCommitTs()), zap.Uint64("commitTs", commitTs))
-}
 
-// ResolveInto appends all messages with CommitTs <= resolve into dst and removes them from the group.
-// ResolveInto copies pointers into dst first, then clears the resolved prefix so Go GC can reclaim
-// resolved messages once downstream is done with them.
-func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage {
-	i := sort.Search(len(g.messages), func(i int) bool {
-		return g.messages[i].GetCommitTs() > resolve
-	})
-	if i == 0 {
-		return dst
+	if outOfOrder {
+		log.Warn("DML events are out of order before flush, sort them",
+			zap.Int32("partition", g.Partition),
+			zap.Int64("tableID", g.tableID),
+			zap.Uint64("resolveTs", resolve),
+			zap.Int("resolved", len(resolved)),
+			zap.Uint64("lastCommitTs", outOfOrderLastTs),
+			zap.Uint64("commitTs", outOfOrderCommitTs))
+		sort.SliceStable(resolved, func(i, j int) bool {
+			return resolved[i].GetCommitTs() < resolved[j].GetCommitTs()
+		})
 	}
 
-	// Copy pointers out first so we can safely clear the group's slice without affecting callers.
-	dst = append(dst, g.messages[:i]...)
-	clear(g.messages[:i])
-	g.messages = g.messages[i:]
+	dst = append(dst, resolved...)
+	clear(original[len(remaining):])
+	g.messages = remaining
 	if len(g.messages) != 0 {
+		firstCommitTs := g.messages[0].GetCommitTs()
 		log.Debug("not all events resolved",
 			zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID),
-			zap.Int("resolved", i), zap.Int("remained", len(g.messages)),
-			zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", g.messages[0].GetCommitTs()))
+			zap.Int("resolved", len(resolved)), zap.Int("remained", len(g.messages)),
+			zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", firstCommitTs))
 	}
 	return dst
 }
 
 // GetAllMessages gets all messages.
 func (g *EventsGroup) GetAllMessages() []*codeccommon.DMLMessage {
-	result := g.messages
-	g.messages = nil
-	return result
+	return g.ResolveInto(math.MaxUint64, nil)
 }
 
 // AppendOrMergeDMLEvent appends a DML event, or merges it into the previous event
diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go
index a8c5fd83ab..da2cbec9c4 100644
--- a/cmd/util/event_group_test.go
+++ b/cmd/util/event_group_test.go
@@ -37,26 +37,26 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D
 	}
 }
 
-func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) {
-	// Scenario: A consumer resolves a prefix of events by watermark/commit-ts and appends them
-	// into a downstream batch slice. We must clear the resolved prefix in the group's backing
-	// array to avoid retaining already-flushed events and causing unbounded memory growth.
+func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) {
+	// Scenario: A consumer resolves events by watermark/commit-ts and appends them into a downstream
+	// batch slice. We must clear resolved messages in the group's backing array to avoid retaining
+	// already-flushed events and causing unbounded memory growth.
 	//
 	// Steps:
 	//  1. Append 3 events with increasing CommitTs.
 	//  2. Call ResolveInto with resolve=2 and a nil dst.
 	//  3. Verify (a) returned events are correct, (b) group keeps only the remaining event,
-	//     (c) the resolved prefix in the original backing slice is cleared (nil'd).
+	//     (c) resolved messages in the original backing slice are cleared (nil'd).
 	group := NewEventsGroup(0, 1)
 	m1 := newTestDMLMessage(1)
 	m2 := newTestDMLMessage(2)
 	m3 := newTestDMLMessage(3)
-	group.AppendMessage(m1, false)
-	group.AppendMessage(m2, false)
-	group.AppendMessage(m3, false)
+	group.AppendMessage(m1)
+	group.AppendMessage(m2)
+	group.AppendMessage(m3)
 
 	// Keep a reference to the original slice header so we can validate that ResolveInto clears
-	// the resolved prefix in-place (this is what prevents GC retention of flushed events).
+	// resolved messages in-place (this is what prevents GC retention of flushed events).
 	original := group.messages
 
 	var dst []*codeccommon.DMLMessage
@@ -69,11 +69,11 @@ func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) {
 	require.Len(t, group.messages, 1)
 	require.Same(t, m3, group.messages[0])
 
-	// The resolved prefix must be nil so the group doesn't keep flushed events alive via its
-	// backing array (classic Go slice memory retention pitfall).
-	require.Nil(t, original[0])
+	// The unresolved event is compacted to the front, and the tail is cleared so the group
+	// doesn't keep flushed events alive via its backing array.
+	require.Same(t, m3, original[0])
 	require.Nil(t, original[1])
-	require.Same(t, m3, original[2])
+	require.Nil(t, original[2])
 }
 
 func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) {
@@ -82,8 +82,8 @@ func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) {
 	group := NewEventsGroup(0, 1)
 	m1 := newTestDMLMessage(10)
 	m2 := newTestDMLMessage(20)
-	group.AppendMessage(m1, false)
-	group.AppendMessage(m2, false)
+	group.AppendMessage(m1)
+	group.AppendMessage(m2)
 
 	original := group.messages
 	dst := make([]*codeccommon.DMLMessage, 0, 1)
@@ -105,8 +105,8 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) {
 	group := NewEventsGroup(0, 1)
 	m1 := newTestDMLMessage(1)
 	m2 := newTestDMLMessage(2)
-	group.AppendMessage(m1, false)
-	group.AppendMessage(m2, false)
+	group.AppendMessage(m1)
+	group.AppendMessage(m2)
 
 	original := group.messages
 	var dst []*codeccommon.DMLMessage
@@ -121,6 +121,67 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) {
 	require.Nil(t, original[1])
 }
 
+func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) {
+	group := NewEventsGroup(0, 1)
+	m1 := newTestDMLMessage(20)
+	m2 := newTestDMLMessage(10)
+	m3 := newTestDMLMessage(30)
+	group.AppendMessage(m1)
+	group.AppendMessage(m2)
+	group.AppendMessage(m3)
+
+	original := group.messages
+	var dst []*codeccommon.DMLMessage
+	dst = group.ResolveInto(25, dst)
+
+	require.Len(t, dst, 2)
+	require.Same(t, m2, dst[0])
+	require.Same(t, m1, dst[1])
+
+	require.Len(t, group.messages, 1)
+	require.Same(t, m3, group.messages[0])
+	require.Same(t, m3, original[0])
+	require.Nil(t, original[1])
+	require.Nil(t, original[2])
+}
+
+func TestEventsGroupResolveIntoKeepsSameCommitTsStable(t *testing.T) {
+	group := NewEventsGroup(0, 1)
+	m1 := newTestDMLMessage(20)
+	m2 := newTestDMLMessage(10)
+	m3 := newTestDMLMessage(20)
+	group.AppendMessage(m1)
+	group.AppendMessage(m2)
+	group.AppendMessage(m3)
+
+	var dst []*codeccommon.DMLMessage
+	dst = group.ResolveInto(20, dst)
+
+	require.Len(t, dst, 3)
+	require.Same(t, m2, dst[0])
+	require.Same(t, m1, dst[1])
+	require.Same(t, m3, dst[2])
+	require.Empty(t, group.messages)
+}
+
+func TestEventsGroupGetAllMessagesSortsOutOfOrderMessages(t *testing.T) {
+	group := NewEventsGroup(0, 1)
+	m1 := newTestDMLMessage(20)
+	m2 := newTestDMLMessage(10)
+	m3 := newTestDMLMessage(30)
+	group.AppendMessage(m1)
+	group.AppendMessage(m2)
+	group.AppendMessage(m3)
+
+	messages := group.GetAllMessages()
+
+	require.Len(t, messages, 3)
+	require.Same(t, m2, messages[0])
+	require.Same(t, m1, messages[1])
+	require.Same(t, m3, messages[2])
+	require.Empty(t, group.messages)
+}
+
 func TestAppendOrMergeDMLEventMergesSameCommitTs(t *testing.T) {
 	var flushed []int
 	e1 := newTestDMLEvent(10, common.RowTypeInsert)