diff --git a/search_api.admin.css b/search_api.admin.css
index c64faf4..0c49e05 100644
--- a/search_api.admin.css
+++ b/search_api.admin.css
@@ -30,10 +30,6 @@ div.search-api-edit-menu.collapsed {
   display: none;
 }
 
-input.search-api-cron-limit {
-  text-align: right;
-}
-
 .search-api-alter-add-aggregation-fields,
 .search-api-checkboxes-list {
   max-height: 12em;
diff --git a/search_api.admin.inc b/search_api.admin.inc
index 865cb4d..e38dc5b 100644
--- a/search_api.admin.inc
+++ b/search_api.admin.inc
@@ -847,37 +847,35 @@ function search_api_admin_index_status_form(array $form, array &$form_state, Sea
       $form['index'] = array(
         '#type' => 'fieldset',
         '#title' => t('Index now'),
-        '#attributes' => array('class' => array('container-inline')),
         '#collapsible' => TRUE,
       );
-      $limit = empty($index->options['cron_limit']) ? SEARCH_API_DEFAULT_CRON_LIMIT : $index->options['cron_limit'];
-      if ($limit > 0) {
-        $form['index']['limit'] = array(
-          '#type' => 'textfield',
-          '#title' => t('Number of batches:'),
-          '#default_value' => -1,
-          '#size' => 4,
-          '#attributes' => array('class' => array('search-api-cron-limit')),
-        );
-        $desc = format_plural($limit,
-              'Each batch contains 1 item. Set to -1 for no limit.',
-              'Each batch contains @count items. Set to -1 for no limit.');
-      }
-      else {
-        $form['index']['limit'] = array(
-          '#type' => 'value',
-          '#value' => -1,
-        );
-        $desc = t("This will immediately index all items that haven't been indexed in their latest version yet.");
-      }
+      $form['index']['settings'] = array(
+        '#type' => 'fieldset',
+        '#title' => t('Advanced settings'),
+        '#collapsible' => TRUE,
+        '#collapsed' => TRUE,
+      );
+      $form['index']['settings']['limit'] = array(
+        '#type' => 'textfield',
+        '#title' => t('Number of items to index'),
+        '#default_value' => -1,
+        '#size' => 4,
+        '#attributes' => array('class' => array('search-api-limit')),
+        '#description' => t('Number of items to index. Set to -1 for all items.'),
+      );
+      $batch_size = empty($index->options['cron_limit']) ? SEARCH_API_DEFAULT_CRON_LIMIT : $index->options['cron_limit'];
+      $form['index']['settings']['batch_size'] = array(
+        '#type' => 'textfield',
+        '#title' => t('Number of items per batch run'),
+        '#default_value' => $batch_size,
+        '#size' => 4,
+        '#attributes' => array('class' => array('search-api-batch-size')),
+        '#description' => t('Number of items per batch run. Set to -1 for all items at once (not recommended). Defaults to the cron batch size of the index.'),
+      );
       $form['index']['button'] = array(
         '#type' => 'submit',
         '#value' => t('Index now'),
       );
-      $form['index']['description'] = array(
-        '#type' => 'item',
-        '#description' => '<br />' . $desc,
-      );
       $form['index']['total'] = array(
         '#type' => 'value',
         '#value' => $status['total'],
@@ -951,24 +949,8 @@ function search_api_admin_index_status_form_submit(array $form, array &$form_sta
       $redirect = $pre . '/disable';
       break;
     case t('Index now'):
-      $limit = $values['limit'];
-      $batch_size = empty($index->options['cron_limit']) ? SEARCH_API_DEFAULT_CRON_LIMIT : $index->options['cron_limit'];
-      // @todo Batch API?
-      try {
-        $ret = search_api_index_items($index, ($limit < 0 || $batch_size < 0) ? -1 : $limit * $batch_size);
-      }
-      catch (SearchApiException $e) {
-        $ret = FALSE;
-        watchdog_exception('search_api', $e);
-      }
-      if ($ret) {
-        drupal_set_message(format_plural($ret, 'Successfully indexed 1 item.', 'Successfully indexed @count items.'));
-        if (($limit > 0 && $ret < min(array($limit, $values['remaining']))) || ($limit < 0 && $ret < $values['remaining'])) {
-          drupal_set_message(t("Some items couldn't be indexed. Check the logs for details."), 'warning');
-        }
-      }
-      else {
-        drupal_set_message(t("Couldn't index items. Check the logs for details."), 'error');
+      if (!_search_api_batch_indexing_create($index, $values['batch_size'], $values['limit'], $values['remaining'])) {
+        drupal_set_message(t("Couldn't create a batch, please check the batch_sise and limit."), 'warning');
       }
       $redirect = $pre . '/status';
       break;
diff --git a/search_api.drush.inc b/search_api.drush.inc
index 9a02106..bf27b15 100644
--- a/search_api.drush.inc
+++ b/search_api.drush.inc
@@ -43,11 +43,13 @@ function search_api_drush_command() {
       'drush sapi-i' => dt('Alias to index items for all enabled indexes.'),
       'drush sapi-i 1' => dt('Index items for the index with the ID !id.', array('!id' => 1)),
       'drush sapi-i default_node_index' => dt('Index items for the index with the machine name !name.', array('!name' => 'default_node_index')),
-      'drush sapi-i 1 100' => dt('Index a maximum number of !limit items for the index with the ID !id.', array('!limit' => 100, '!id' => 1)),
+      'drush sapi-i 1 100' => dt("Index a maximum number of !limit items (index's cron batch size items per batch run) for the index with the ID !id.", array('!limit' => 100, '!id' => 1)),
+      'drush sapi-i 1 100 10' => dt("Index a maximum number of !limit items (!batch_size items per batch run) for the index with the ID !id.", array('!limit' => 100, '!batch_size' => 10, '!id' => 1)),
     ),
     'arguments' => array(
       'index_id' => dt('The numeric ID or machine name of an index.'),
-      'limit' => dt("The number of items to index. Use 0 to index all items. Defaults to the index's cron batch size."),
+      'limit' => dt("The number of items to index (index's cron batch size items per run). Set to 0 to index all items. Defaults to 0 (index all)."),
+      'batch_size' => dt("The number of items to index per batch run. Set to 0 to index all items at once. Defaults to the index's cron batch size."),
     ),
     'aliases' => array('sapi-i'),
   );
@@ -161,8 +163,15 @@ function drush_search_api_status($index_id = NULL) {
 
 /**
  * Index items.
+ *
+ * @param string|integer $index_id
+ *   The index name or id for which items should be indexed.
+ * @param integer $limit
+ *   Maximum number of items to index.
+ * @param integer $batch_size
+ *   Number of items to index per batch.
  */
-function drush_search_api_index($index_id = NULL, $limit = NULL) {
+function drush_search_api_index($index_id = NULL, $limit = NULL, $batch_size = NULL) {
   if (search_api_drush_static(__FUNCTION__)) {
     return;
   }
@@ -170,18 +179,65 @@ function drush_search_api_index($index_id = NULL, $limit = NULL) {
   if (empty($indexes)) {
     return;
   }
-  $limit_string = $limit;
   foreach ($indexes as $index) {
-    if (is_null($limit)) {
-      $limit = $index->options['cron_limit'];
-      $limit_string = $limit;
+    // Get the number of remaing items to index.
+    $datasource = $index->datasource();
+    $index_status = $datasource->getIndexStatus($index);
+    $remaining = $index_status['total'] - $index_status['indexed'];
+
+    // Get the number of items to index per batch run.
+    if (!isset($batch_size)) {
+      $batch_size = empty($index->options['cron_limit']) ? SEARCH_API_DEFAULT_CRON_LIMIT : $index->options['cron_limit'];
+    }
+    elseif ($batch_size <= 0) {
+      $batch_size = $remaining;
+    }
+
+    // Get the number items to index.
+    if (!isset($limit) || !is_int($limit += 0) || $limit <= 0) {
+      $limit = $remaining;
+    }
+
+    drush_log(dt("Indexing a maximum number of !limit items (!batch_size items per batch run) for the index !index.", array('!index' => $index->name, '!limit' => $limit, '!batch_size' => $batch_size)), 'ok');
+
+    // Create the batch.
+    if (!_search_api_batch_indexing_create($index, $batch_size, $limit, $remaining, TRUE)) {
+      drush_log(dt("Couldn't create a batch, please check the batch_sise and limit parameters."), 'error');
+    }
+    else {
+      // Launch the batch process.
+      drush_backend_batch_process();
     }
-    elseif ((int) $limit <= 0) {
-      $limit = -1;
-      $limit_string = t('all');
+  }
+}
+
+/**
+ * Copy of formal_plural that works with drush as 't' may not be available.
+ */
+function _search_api_drush_format_plural($count, $singular, $plural, array $args = array(), array $options = array()) {
+  $args['@count'] = $count;
+  if ($count == 1) {
+    return dt($singular, $args, $options);
+  }
+
+  // Get the plural index through the gettext formula.
+  $index = (function_exists('locale_get_plural')) ? locale_get_plural($count, isset($options['langcode']) ? $options['langcode'] : NULL) : -1;
+  // If the index cannot be computed, use the plural as a fallback (which
+  // allows for most flexiblity with the replaceable @count value).
+  if ($index < 0) {
+    return dt($plural, $args, $options);
+  }
+  else {
+    switch ($index) {
+      case "0":
+        return dt($singular, $args, $options);
+      case "1":
+        return dt($plural, $args, $options);
+      default:
+        unset($args['@count']);
+        $args['@count[' . $index . ']'] = $count;
+        return dt(strtr($plural, array('@count' => '@count[' . $index . ']')), $args, $options);
     }
-    drush_print(dt('Indexing a maximum number of !limit items for index !index.', array('!index' => $index->name, '!limit' => $limit_string)));
-    search_api_index_items($index, $limit);
   }
 }
 
diff --git a/search_api.module b/search_api.module
index 1014ff4..3b7f771 100644
--- a/search_api.module
+++ b/search_api.module
@@ -1035,6 +1035,16 @@ function search_api_index_specific_items(SearchApiIndex $index, array $ids) {
   $indexed = $items ? $index->index($cloned_items) : array();
   if ($indexed) {
     search_api_track_item_indexed($index, $indexed);
+    // If some items could not be indexed, we don't want to try re-indexing
+    // them right away, so we mark them as "freshly" changed. Sadly, there is
+    // no better way than to mark them as indexed first...
+    if (count($indexed) < count($ids)) {
+      // Believe it or not but this is actually quite faster than the equivalent
+      // $diff = array_diff($ids, $indexed);
+      $diff = array_keys(array_diff_key(array_flip($ids), array_flip($indexed)));
+      $index->datasource()->trackItemIndexed($diff, $index);
+      $index->datasource()->trackItemChange($diff, array($index));
+    }
   }
   return $indexed;
 }
@@ -2175,3 +2185,144 @@ function _search_api_convert_custom_type($callback, $value, $original_type, $typ
   }
   return $values;
 }
+
+/**
+ * Create and set a batch for indexing items.
+ *
+ * @param SearchApiIndex $index
+ *   The index for which items should be indexed.
+ * @param $batch_size
+ *   Number of items to index per batch.
+ * @param $limit
+ *   Maximum number of items to index.
+ * @param $remaining
+ *   Remaining items to index.
+ * @param $drush
+ *   Boolean specifying whether this was called from drush or not.
+ */
+function _search_api_batch_indexing_create(SearchApiIndex $index, $batch_size, $limit, $remaining, $drush = FALSE) {
+  if ($limit !== 0 && $batch_size !== 0) {
+    $t = !empty($drush) ? 'dt' : 't';
+
+    if ($limit < 0 || $limit > $remaining) {
+      $limit = $remaining;
+    }
+    if ($batch_size < 0) {
+      $batch_size = $remaining;
+    }
+    $batch = array(
+      'title' => $t('Indexing items'),
+      'operations' => array(
+        array('_search_api_batch_indexing_callback', array($index, $batch_size, $limit, $drush)),
+      ),
+      'progress_message' => $t('Completed about @percentage% of the indexing operation.'),
+      'finished' => '_search_api_batch_indexing_finished',
+      'file' => drupal_get_path('module', 'search_api') . '/search_api.module',
+    );
+    batch_set($batch);
+    return TRUE;
+  }
+  return FALSE;
+}
+
+/**
+ * Batch API callback for the indexing functionality.
+ *
+ * @param SearchApiIndex $index
+ *   The index for which items should be indexed.
+ * @param integer $batch_size
+ *   Number of items to index per batch.
+ * @param integer $limit
+ *   Maximum number of items to index.
+ * @param boolean $drush
+ *   Boolean specifying whether this was called from drush or not.
+ * @param array $context
+ *   The batch context.
+ */
+function _search_api_batch_indexing_callback(SearchApiIndex $index, $batch_size, $limit, $drush = FALSE, array &$context) {
+  // Persistent data among batch runs.
+  if (!isset($context['sandbox']['limit'])) {
+    $context['sandbox']['limit'] = $limit;
+    $context['sandbox']['batch_size'] = $batch_size;
+    $context['sandbox']['progress'] = 0;
+  }
+
+  // Persistent data for results.
+  if (!isset($context['results']['indexed'])) {
+    $context['results']['indexed'] = 0;
+    $context['results']['not indexed'] = 0;
+    $context['results']['drush'] = $drush;
+  }
+
+  // Number of items to index for this run.
+  $to_index = min($context['sandbox']['limit'] - $context['sandbox']['progress'], $context['sandbox']['batch_size']);
+
+  // Index the items.
+  $indexed = search_api_index_items($index, $to_index);
+  $context['results']['indexed'] += $indexed;
+
+  // Display progress message.
+  if ($indexed > 0) {
+    $format_plural = $context['results']['drush'] === TRUE ? '_search_api_drush_format_plural' : 'format_plural';
+    $context['message'] = $format_plural($context['results']['indexed'], 'Successfully indexed 1 item.', 'Successfully indexed @count items.');
+  }
+
+  // Some items couldn't be indexed.
+  if ($indexed !== $to_index) {
+    $context['results']['not indexed'] += $to_index - $indexed;
+  }
+
+  $context['sandbox']['progress'] += $to_index;
+
+  // Everything has been indexed.
+  if ($indexed === 0 || $context['sandbox']['progress'] >= $context['sandbox']['limit']) {
+    $context['finished'] = 1;
+  }
+  else {
+    $context['finished'] = $context['sandbox']['progress'] / $context['sandbox']['limit'];
+  }
+}
+
+/**
+ * Batch API finishing callback for the indexing functionality.
+ *
+ * @param boolean $success
+ *   Result of the batch operation.
+ * @param array $results
+ *   Results.
+ * @param array $operations
+ *   Remaining batch operation to process.
+ */
+function _search_api_batch_indexing_finished($success, $results, $operations) {
+  // Check if called from drush.
+  if (!empty($results['drush'])) {
+    $drupal_set_message = 'drush_log';
+    $format_plural = '_search_api_drush_format_plural';
+    $t = 'dt';
+    $success_message = 'success';
+  }
+  else {
+    $drupal_set_message = 'drupal_set_message';
+    $format_plural = 'format_plural';
+    $t = 't';
+    $success_message = 'status';
+  }
+
+  // Display result messages.
+  if ($success) {
+    if (!empty($results['indexed'])) {
+      $drupal_set_message($format_plural($results['indexed'], 'Successfully indexed 1 item.', 'Successfully indexed @count items.'), $success_message);
+
+      if (!empty($results['not indexed'])) {
+       $drupal_set_message($format_plural($results['not indexed'], '1 item could not be indexed. Check the logs for details.', '@count items could not be indexed. Check the logs for details.'), 'warning');
+      }
+    }
+    else {
+      $drupal_set_message($t("Couldn't index items. Check the logs for details."), 'error');
+    }
+  }
+  else {
+    $drupal_set_message($t("An error occurred while trying to index items. Check the logs for details."), 'error');
+  }
+
+}
diff --git a/search_api.test b/search_api.test
index c3cfbc1..e65cc7a 100644
--- a/search_api.test
+++ b/search_api.test
@@ -296,26 +296,34 @@ class SearchApiWebTest extends DrupalWebTestCase {
     $this->assertText(t('Clear index'), t('"Clear index" button found.'));
     $this->assertNoText(t('Re-index content'), t('"Re-index" button not found.'));
 
+    // Here we test the indexing + the warning message when some items
+    // can not be indexed.
+    // The server refuses (for test purpose) to index items with IDs that are
+    // multiples of 8 unless the "search_api_test_index_all" variable is set.
     $values = array(
       'limit' => 8,
     );
     $this->drupalPost(NULL, $values, t('Index now'));
     $this->assertText(t('Successfully indexed @count items.', array('@count' => 7)));
-    $this->assertText(t("Some items couldn't be indexed. Check the logs for details."), t('Index errors warning is displayed.'));
+    $this->assertText(t('1 item could not be indexed. Check the logs for details.'), t('Index errors warning is displayed.'));
     $this->assertNoText(t("Couldn't index items. Check the logs for details."), t("Index error isn't displayed."));
     $this->assertText(t('About @percentage% of all items have been indexed in their latest version (@indexed / @total).', array('@indexed' => 7, '@total' => 10, '@percentage' => 70)), t('!field displayed.', array('!field' => t('Correct index status'))));
     $this->assertText(t('Re-indexing'), t('"Re-index" button found.'));
 
+    // Here we're testing the error message when no item could be indexed.
+    // The item with ID 8 is still not indexed.
     $values = array(
       'limit' => 1,
     );
     $this->drupalPost(NULL, $values, t('Index now'));
     $this->assertNoPattern('/' . str_replace('144', '-?\d*', t('Successfully indexed @count items.', array('@count' => 144))) . '/', t('No items could be indexed.'));
-    $this->assertNoText(t("Some items couldn't be indexed. Check the logs for details."), t("Index errors warning isn't displayed."));
+    $this->assertNoText(t('1 item could not be indexed. Check the logs for details.'), t("Index errors warning isn't displayed."));
     $this->assertText(t("Couldn't index items. Check the logs for details."), t('Index error is displayed.'));
 
+    // Here we test the indexing of all the remaining items.
+    variable_set('search_api_test_index_all', TRUE);
     $values = array(
-      'limit' => 100,
+      'limit' => -1,
     );
     $this->drupalPost(NULL, $values, t('Index now'));
     $this->assertText(t('Successfully indexed @count items.', array('@count' => 3)));
diff --git a/tests/search_api_test.module b/tests/search_api_test.module
index decddf0..998598e 100644
--- a/tests/search_api_test.module
+++ b/tests/search_api_test.module
@@ -231,14 +231,14 @@ class SearchApiTestService extends SearchApiAbstractService {
   }
 
   public function indexItems(SearchApiIndex $index, array $items) {
-    // Refuse to index items with IDs that are multiples of 7 unless all items are indexed.
-    $count = db_query('SELECT COUNT(*) FROM {search_api_item} WHERE index_id = :index_id AND changed <> 0', array(':index_id' => $index->id))->fetchField();
-    if ($count == count($items)) {
+    // Refuse to index items with IDs that are multiples of 8 unless the
+    // "search_api_test_index_all" variable is set.
+    if (variable_get('search_api_test_index_all', FALSE)) {
       return $this->index($index, array_keys($items));
     }
     $ret = array();
     foreach ($items as $id => $item) {
-      if ($id % 7) {
+      if ($id % 8) {
         $ret[] = $id;
       }
     }