sort events

pingcap · Dec 23, 2024 · 5bedd08 · 5bedd08
1 parent d6b5c81
commit 5bedd08
Showing 1 changed file with 17 additions and 13 deletions.
diff --git a/cmd/kafka-consumer/event_group.go b/cmd/kafka-consumer/event_group.go
@@ -60,19 +60,23 @@ func (g *eventsGroup) Append(row *model.RowChangedEvent, offset kafka.Offset) {
 
 // Resolve will get events where CommitTs is less than resolveTs.
 func (g *eventsGroup) Resolve(resolve uint64, protocol config.Protocol) []*model.RowChangedEvent {
-	switch protocol {
-	case config.ProtocolCanalJSON:
-		sort.Slice(g.events, func(i, j int) bool {
-			return g.events[i].CommitTs < g.events[j].CommitTs
-		})
-	default:
-		if !sort.SliceIsSorted(g.events, func(i, j int) bool {
-			return g.events[i].CommitTs < g.events[j].CommitTs
-		}) {
-			log.Warn("events are not sorted", zap.Int32("partition", g.partition),
-				zap.Int64("tableID", g.tableID), zap.Int("eventCount", len(g.events)))
-		}
-	}
+	//switch protocol {
+	//case config.ProtocolCanalJSON:
+	//	sort.Slice(g.events, func(i, j int) bool {
+	//		return g.events[i].CommitTs < g.events[j].CommitTs
+	//	})
+	//default:
+	//	if !sort.SliceIsSorted(g.events, func(i, j int) bool {
+	//		return g.events[i].CommitTs < g.events[j].CommitTs
+	//	}) {
+	//		log.Warn("events are not sorted", zap.Int32("partition", g.partition),
+	//			zap.Int64("tableID", g.tableID), zap.Int("eventCount", len(g.events)))
+	//	}
+	//}
+
+	sort.Slice(g.events, func(i, j int) bool {
+		return g.events[i].CommitTs < g.events[j].CommitTs
+	})
 
 	i := sort.Search(len(g.events), func(i int) bool {
 		return g.events[i].CommitTs > resolve