sinkv2(ticdc): remove redundant nextToMount in MountedEventIter (#8030)

ref #5928

cdc/processor/sourcemanager/engine/mounted_iter.go

@@ -27,7 +27,6 @@ type MountedEventIter struct {
 	maxBatchSize int
 
 	rawEvents      []rawEvent
-	nextToMount    int
 	nextToEmit     int
 	savedIterError error
 }
@@ -47,54 +46,57 @@ func NewMountedEventIter(
 
 // Next returns the next mounted event.
 func (i *MountedEventIter) Next(ctx context.Context) (event *model.PolymorphicEvent, txnFinished Position, err error) {
-	// Check whether there are events in mounting or not.
-	for idx := i.nextToEmit; idx < i.nextToMount; idx++ {
-		if err = i.rawEvents[idx].event.WaitFinished(ctx); err == nil {
-			event = i.rawEvents[idx].event
-			txnFinished = i.rawEvents[idx].txnFinished
-			i.nextToEmit += 1
+	// There are no events in mounting. Fetch more events and mounting them.
+	// The batch size is determined by `maxBatchSize`.
+	if i.nextToEmit >= len(i.rawEvents) {
+		if err = i.readBatch(ctx); err != nil {
+			return
 		}
-		return
 	}
 
-	// There are no events in mounting. Fetch more events and mounting them.
-	// The batch size is determined by `maxBatchSize`.
-	if i.mg != nil && i.iter != nil {
-		i.nextToMount = 0
-		i.nextToEmit = 0
-		if cap(i.rawEvents) == 0 {
-			i.rawEvents = make([]rawEvent, 0, i.maxBatchSize)
-		} else {
-			i.rawEvents = i.rawEvents[:0]
+	// Check whether there are events in mounting or not.
+	if i.nextToEmit < len(i.rawEvents) {
+		idx := i.nextToEmit
+		if err = i.rawEvents[idx].event.WaitFinished(ctx); err != nil {
+			return
 		}
+		event = i.rawEvents[idx].event
+		txnFinished = i.rawEvents[idx].txnFinished
+		i.nextToEmit += 1
+	}
+	return
+}
+
+func (i *MountedEventIter) readBatch(ctx context.Context) error {
+	if i.mg == nil || i.iter == nil {
+		return nil
+	}
+
+	i.nextToEmit = 0
+	if cap(i.rawEvents) == 0 {
+		i.rawEvents = make([]rawEvent, 0, i.maxBatchSize)
+	} else {
+		i.rawEvents = i.rawEvents[:0]
+	}
 
-		for len(i.rawEvents) < cap(i.rawEvents) {
-			event, txnFinished, err = i.iter.Next()
-			if err != nil {
-				return
-			}
-			if event == nil {
-				i.savedIterError = i.iter.Close()
-				i.iter = nil
-				break
-			}
-			i.rawEvents = append(i.rawEvents, rawEvent{event, txnFinished})
+	for len(i.rawEvents) < cap(i.rawEvents) {
+		event, txnFinished, err := i.iter.Next()
+		if err != nil {
+			return err
 		}
-		for idx := i.nextToMount; idx < len(i.rawEvents); idx++ {
-			i.rawEvents[idx].event.SetUpFinishedCh()
-			if err = i.mg.AddEvent(ctx, i.rawEvents[idx].event); err != nil {
-				i.mg = nil
-				return
-			}
-			i.nextToMount += 1
+		if event == nil {
+			i.savedIterError = i.iter.Close()
+			i.iter = nil
+			break
 		}
-
-		// More events are fetched and in mounting. So re-call this function to wait them.
-		if i.nextToEmit < i.nextToMount {
-			return i.Next(ctx)
+		event.SetUpFinishedCh()
+		if err := i.mg.AddEvent(ctx, event); err != nil {
+			i.mg = nil
+			return err
 		}
+		i.rawEvents = append(i.rawEvents, rawEvent{event, txnFinished})
 	}
-	return
+	return nil
 }
 
 // Close implements sorter.EventIterator.

cdc/processor/sourcemanager/engine/mounted_iter_test.go

@@ -56,14 +56,12 @@ func TestMountedEventIter(t *testing.T) {
 		require.NotNil(t, event)
 		require.Nil(t, err)
 	}
-	require.Equal(t, iter.nextToMount, 3)
 	require.Equal(t, iter.nextToEmit, 3)
 
 	rawIter.repeatItem = func() *model.PolymorphicEvent { return nil }
 	event, _, err := iter.Next(context.Background())
 	require.Nil(t, event)
 	require.Nil(t, err)
-	require.Equal(t, iter.nextToMount, 0)
 	require.Equal(t, iter.nextToEmit, 0)
 	require.Nil(t, iter.iter)
 }