Continue trying to process the batch and retry on UnproccessedItems as
according to the specified BatchWriteRetryStrategy
436354
private FailedBatch doBatchWriteItemWithRetry(
            Map<String, List<WriteRequest>> batch,
            BatchWriteRetryStrategy batchWriteRetryStrategy) {

        BatchWriteItemResult result = null;
        int retries = 0;
        int maxRetries = batchWriteRetryStrategy
                .getMaxRetryOnUnprocessedItems(Collections
                        .unmodifiableMap(batch));

        FailedBatch failedBatch = null;
        Map<String, List<WriteRequest>> pendingItems = batch;

        while (true) {
            try {
                result = db.batchWriteItem(applyBatchOperationUserAgent(
                        new BatchWriteItemRequest().withRequestItems(pendingItems)));
            } catch (Exception e) {
                failedBatch = new FailedBatch();
                failedBatch.setUnprocessedItems(pendingItems);
                failedBatch.setException(e);
                return failedBatch;
            }
            pendingItems = result.getUnprocessedItems();

            if (pendingItems.size() > 0) {

                // return pendingItems as a FailedBatch if we have exceeded max retry
                if (maxRetries >= 0 && retries >= maxRetries) {
                    failedBatch = new FailedBatch();
                    failedBatch.setUnprocessedItems(pendingItems);
                    failedBatch.setException(null);
                    return failedBatch;
                }

                pause(batchWriteRetryStrategy.getDelayBeforeRetryUnprocessedItems(
                        Collections.unmodifiableMap(pendingItems), retries));
                retries++;
            } else {
                break;
            }
        }
        return failedBatch;
    }
811009
public Future<WriteResponse> write (Batch<String> batch, WriteCallback callback) {
    Timer.Context context = writeTimer.time();
    int returnCode = 0;
    LOG.info ("Dispatching batch " + batch.getId());
    recordsAttempted.mark(batch.getRecords().size());
    try {
      String encoded = encodeBatch(batch);
      returnCode = request (encoded);
      WriteResponse<Integer> response = WRITE_RESPONSE_WRAPPER.wrap(returnCode);
      callback.onSuccess(response);
      bytesWritten.mark(encoded.length());
      recordsSuccess.mark(batch.getRecords().size());
    } catch (Exception e) {
      LOG.error("Dispatching batch " + batch.getId() + " failed :" + e.toString());
      callback.onFailure(e);
      recordsFailed.mark(batch.getRecords().size());
    }

    context.close();
    Future<Integer> future = Futures.immediateFuture(returnCode);
    return new WriteResponseFuture<>(future, WRITE_RESPONSE_WRAPPER);
  }
413384
public function nextItem()
    {
        if ($this->batch === null) {
            $this->batch = [];
        }
        
        //If no items in batch, and we have a resumptionToken or need to make initial request...
        if (count($this->batch) == 0 && ($this->resumptionToken or $this->numRequests == 0)) {
            $this->retrieveBatch();
        }

        //if still items in current batch, return one
        if (count($this->batch) > 0) {
            $this->numProcessed++;

            $item = array_shift($this->batch);
            $this->currItem = clone $item;
        } else {
            $this->currItem = false;
        }

        return $this->currItem;
    }
723899
def write_batch(self, batch):
        
        for item in batch:
            for key in item:
                self.aggregated_info['occurrences'][key] += 1
            self.increment_written_items()
            if self.items_limit and self.items_limit == self.get_metadata('items_count'):
                raise ItemsLimitReached('Finishing job after items_limit reached: {} items written.'
                                        .format(self.get_metadata('items_count')))
        self.logger.debug('Wrote items')
671928
private void initFlushOperation(String bulkLoggingID, boolean retryOperation, long retriedDocs, long waitTime) {
        if (retryOperation) {
            if (waitTime > 0L) {
                debugLog(bulkLoggingID, "Retrying [%d] entries after backing off for [%s] ms",
                        retriedDocs, TimeValue.timeValueMillis(waitTime));
                try {
                    Thread.sleep(waitTime);
                } catch (InterruptedException e) {
                    debugLog(bulkLoggingID, "Thread interrupted - giving up on retrying...");
                    throw new EsHadoopException("Thread interrupted - giving up on retrying...", e);
                }
            } else {
                debugLog(bulkLoggingID, "Retrying [%d] entries immediately (without backoff)", retriedDocs);
            }
        } else {
            debugLog(bulkLoggingID, "Sending batch of [%d] bytes/[%s] entries", data.length(), dataEntries);
        }
    }
355177
def resend_unprocessed(self):
        
        LOG.info("Re-sending %d unprocessed items.", len(self._unprocessed))

        while self._unprocessed:
            to_resend = self._unprocessed[:MAX_WRITE_BATCH]
            self._unprocessed = self._unprocessed[MAX_WRITE_BATCH:]
            LOG.info("Sending %d items", len(to_resend))
            self._write(to_resend)
            LOG.info("%d unprocessed items left", len(self._unprocessed))
534401
def next_task(self, item, raise_exceptions=None, **kwargs):
        
        filename = os.path.basename(item)
        batch = self.get_batch(filename)
        tx_deserializer = self.tx_deserializer_cls(
            allow_self=self.allow_self, override_role=self.override_role
        )
        try:
            tx_deserializer.deserialize_transactions(
                transactions=batch.saved_transactions
            )
        except (DeserializationError, TransactionDeserializerError) as e:
            raise TransactionsFileQueueError(e) from e
        else:
            batch.close()
            self.archive(filename)
788405
def _batch_write_item(self, items):
        
        kwargs = {
            'RequestItems': {
                self.tablename: items,
            },
            'ReturnConsumedCapacity': self.return_capacity,
            'ReturnItemCollectionMetrics': self.return_item_collection_metrics,
        }
        return self.connection.call('batch_write_item', **kwargs)
1162378
def batch!(operations, request_options = {})
      res = batch(operations, request_options)
      res['taskID'].each do |index, taskID|
        wait_task(index, taskID, WAIT_TASK_DEFAULT_TIME_BEFORE_RETRY, request_options)
      end
    end
680754
def process_request(self, request):
        """"""
        batcher = PrioritizedBatcher.global_instance()
        if batcher.is_started:
            # This can happen in old-style middleware if consequent middleware
            # raises exception and thus `process_response` is not called.
            # Described under 3rd point of differences:
            # https://docs.djangoproject.com/en/1.11/topics/http/middleware/#upgrading-pre-django-1-10-style-middleware
            batcher.rollback()
            logger.warning(
                "Uncommited batcher transaction from previous request was rollbacked."
            )

        batcher.start()
1093522
@Override
  public synchronized void write(final Collection<BatchPoints> collection) {
    // empty the cached data first
    ListIterator<BatchPoints> batchQueueIterator = batchQueue.listIterator();
    while (batchQueueIterator.hasNext()) {
      BatchPoints entry = batchQueueIterator.next();
      WriteResult result = tryToWrite(entry);
      if (result.outcome == WriteResultOutcome.WRITTEN
              || result.outcome == WriteResultOutcome.FAILED_RETRY_IMPOSSIBLE) {
        batchQueueIterator.remove();
        usedRetryBufferCapacity -= entry.getPoints().size();
        // we are throwing out data, notify the client
        if (result.outcome == WriteResultOutcome.FAILED_RETRY_IMPOSSIBLE) {
          exceptionHandler.accept(entry.getPoints(), result.throwable);
        }
      } else {
        // we cannot send more data otherwise we would write them in different
        // order than in which were submitted
        for (BatchPoints batchPoints : collection) {
          addToBatchQueue(batchPoints);
        }
        return;
      }
    }
    // write the last given batch last so that duplicate data points get overwritten correctly
    Iterator<BatchPoints> collectionIterator = collection.iterator();
    while (collectionIterator.hasNext()) {
      BatchPoints batchPoints = collectionIterator.next();
      WriteResult result = tryToWrite(batchPoints);
      switch (result.outcome) {
        case FAILED_RETRY_POSSIBLE:
          addToBatchQueue(batchPoints);
          while (collectionIterator.hasNext()) {
            addToBatchQueue(collectionIterator.next());
          }
          break;
        case FAILED_RETRY_IMPOSSIBLE:
          exceptionHandler.accept(batchPoints.getPoints(), result.throwable);
          break;
        default:

      }
    }
  }
528790
@Override
    public void run() {

        if (this.state != null && this.state.cancel) {
            // we were already cancelled, don't run, but still post completion
            this.state.documentCounter = 0;
            this.state.batchCounter = 0;
            runComplete(null);
            return;
        }
        // reset internal state
        this.state = new State();

        Throwable errorInfo = null;

        try {
            this.useBulkGet = sourceDb.isBulkSupported();
            replicate();

        } catch (ExecutionException ex) {
            logger.log(Level.SEVERE, String.format("Batch %s ended with error:", this.state
                    .batchCounter), ex);
            errorInfo = ex.getCause();
        } catch (Throwable e) {
            logger.log(Level.SEVERE, String.format("Batch %s ended with error:", this.state
                    .batchCounter), e);
            errorInfo = e;
        }

        runComplete(errorInfo);
    }
943613
private void postProcess(ProcessMethodCallback callback, TransactionContext txContext,
                           InputDatum input, ProcessMethod.ProcessResult result) {
    InputContext inputContext = input.getInputContext();
    Throwable failureCause = null;
    FailureReason.Type failureType = FailureReason.Type.IO_ERROR;
    try {
      if (result.isSuccess()) {
        // If it is a retry input, force the dequeued entries into current transaction.
        if (input.getRetry() > 0) {
          input.reclaim();
        }
        txContext.finish();
      } else {
        failureCause = result.getCause();
        failureType = FailureReason.Type.USER;
        txContext.abort();
      }
    } catch (Throwable e) {
      LOG.error("Transaction operation failed: {}", e.getMessage(), e);
      failureType = FailureReason.Type.IO_ERROR;
      if (failureCause == null) {
        failureCause = e;
      }
      try {
        if (result.isSuccess()) {
          txContext.abort();
        }
      } catch (Throwable ex) {
        LOG.error("Fail to abort transaction: {}", inputContext, ex);
      }
    }

    try {
      if (failureCause == null) {
        callback.onSuccess(result.getEvent(), inputContext);
      } else {
        callback.onFailure(result.getEvent(), inputContext,
                           new FailureReason(failureType, failureCause.getMessage(), failureCause),
                           createInputAcknowledger(input));
      }
    } catch (Throwable t) {
      LOG.error("Failed to invoke callback.", t);
    }
  }
1138160
void afterWrite(Runnable task) {
    if (buffersWrites()) {
      for (int i = 0; i < WRITE_BUFFER_RETRIES; i++) {
        if (writeBuffer().offer(task)) {
          scheduleAfterWrite();
          return;
        }
        scheduleDrainBuffers();
      }

      // The maintenance task may be scheduled but not running due to all of the executor's threads
      // being busy. If all of the threads are writing into the cache then no progress can be made
      // without assistance.
      try {
        performCleanUp(task);
      } catch (RuntimeException e) {
        logger.log(Level.SEVERE, "Exception thrown when performing the maintenance task", e);
      }
    } else {
      scheduleAfterWrite();
    }
  }
263343
def _complete_batch_send(self, resp):
        
        self._batch_send_d = None
        self._req_attempts = 0
        self._retry_interval = self._init_retry_interval
        if isinstance(resp, Failure) and not resp.check(tid_CancelledError,
                                                        CancelledError):
            log.error("Failure detected in _complete_batch_send: %r\n%r",
                      resp, resp.getTraceback())
        return
1170996
function proceedWithBackup() {
    if (options.resume) {
      // pick up from existing log file from previous run
      downloadRemainingBatches(options.log, db, ee, start, batchesPerDownloadSession, options.parallelism);
    } else {
      // create new log file and process
      spoolchanges(db, options.log, options.bufferSize, ee, function(err) {
        if (err) {
          ee.emit('error', err);
        } else {
          downloadRemainingBatches(options.log, db, ee, start, batchesPerDownloadSession, options.parallelism);
        }
      });
    }
  }
360469
@VisibleForTesting
    protected void executionComplete(Throwable exception) {
        Collection<QueueItem> toFail = null;
        Throwable failEx = null;
        synchronized (this.stateLock) {
            // Release the spot occupied by this item's execution.
            this.activeCount--;
            if (exception != null && !this.closed) {
                // Need to fail all future items and close to prevent new items from being processed.
                failEx = new ProcessingException("A previous item failed to commit. Cannot process new items.", exception);
                toFail = new ArrayList<>(this.pendingItems);
                this.pendingItems.clear();
                this.closed = true;
            }

            if (this.emptyNotifier != null && this.activeCount == 0 && this.pendingItems.isEmpty()) {
                // We were asked to notify when we were empty.
                this.emptyNotifier.release();
                this.emptyNotifier = null;
            }
        }

        if (toFail != null) {
            for (QueueItem q : toFail) {
                q.result.completeExceptionally(failEx);
            }

            return;
        }

        // We need to ensure the items are still executed in order. Once out of the main sync block, it is possible that
        // this callback may be invoked concurrently after various completions. As such, we need a special handler for
        // these, using its own synchronization, different from the main one, so that we don't hold that stateLock for too long.
        synchronized (this.processingLock) {
            while (true) {
                QueueItem toProcess;
                synchronized (this.stateLock) {
                    if (hasCapacity() && !this.pendingItems.isEmpty()) {
                        // We have spare capacity and we have something to process. Dequeue it and reserve the spot.
                        toProcess = this.pendingItems.pollFirst();
                        this.activeCount++;
                    } else {
                        // No capacity left or no more pending items.
                        break;
                    }
                }

                Futures.completeAfter(() -> processInternal(toProcess.data), toProcess.result);
            }
        }
    }