refactor

Author: raulb

source/logrepl/combined.go

@@ -119,28 +119,21 @@ func (c *CombinedIterator) NextN(ctx context.Context, n int) ([]opencdc.Record,
 		return nil, fmt.Errorf("n must be greater than 0, got %d", n)
 	}
 
-	// Check if the active iterator already implements NextN (like the CDCIterator)
-	if nextN, ok := c.activeIterator.(interface {
-		NextN(context.Context, int) ([]opencdc.Record, error)
-	}); ok {
-		records, err := nextN.NextN(ctx, n)
-		if err != nil {
-			// Snapshot iterator is done, handover to CDC iterator
-			if errors.Is(err, snapshot.ErrIteratorDone) {
-				if err := c.useCDCIterator(ctx); err != nil {
-					return nil, err
-				}
-				sdk.Logger(ctx).Debug().Msg("Snapshot completed, switching to CDC mode")
-
-				// Retry with new iterator
-				return c.NextN(ctx, n)
+	records, err := c.activeIterator.NextN(ctx, n)
+	if err != nil {
+		// Snapshot iterator is done, handover to CDC iterator
+		if errors.Is(err, snapshot.ErrIteratorDone) {
+			if err := c.useCDCIterator(ctx); err != nil {
+				return nil, err
 			}
-			return nil, fmt.Errorf("failed to fetch records in batch: %w", err)
+			sdk.Logger(ctx).Debug().Msg("Snapshot completed, switching to CDC mode")
+
+			// Retry with new iterator
+			return c.NextN(ctx, n)
 		}
-		return records, nil
+		return nil, fmt.Errorf("failed to fetch records in batch: %w", err)
 	}
-
-	return nil, nil
+	return records, nil
 }
 
 func (c *CombinedIterator) Ack(ctx context.Context, p opencdc.Position) error {