Hand off a batch to the DbHarness layer, retrying on FullQueueException.
517104
private Future<?> runBatch(Batch<T> batch) throws InterruptedException {
        while (true) {
            try {
                runBatchMeter.mark();
                if (perRollupMetrics) {
                    batch.getMap().forEach((addr, op) -> {
                        if (addr.getSourceRollup().isPresent()) {
                            updateRollupHistogram(rollupWriteSize, addr.getSourceRollup().get(), "rollupWriteSize", op);
                        }
                    });
                }

                return db.runBatchAsync(batch, flushErrorHandler);
            } catch (FullQueueException e) {
                asyncQueueBackoffMeter.mark();

                // Sleeping and retrying like this means batches may be flushed out of order
                log.debug("Async queue is full, retrying soon");
                Thread.sleep(100);
            }
        }
    }
91602
private void doStartBatch(String batchID) {
    if (currentBatch != null && currentBatch.id().equals(batchID)) {
      currentBatch.handleStart();
    }
  }
1221913
@SuppressWarnings("unchecked")
    private CompletionStage<List<V>> invokeMapBatchLoader(List<K> keys, BatchLoaderEnvironment environment) {
        CompletionStage<Map<K, V>> loadResult;
        Set<K> setOfKeys = new LinkedHashSet<>(keys);
        if (batchLoadFunction instanceof MappedBatchLoaderWithContext) {
            loadResult = ((MappedBatchLoaderWithContext<K, V>) batchLoadFunction).load(setOfKeys, environment);
        } else {
            loadResult = ((MappedBatchLoader<K, V>) batchLoadFunction).load(setOfKeys);
        }
        CompletionStage<Map<K, V>> mapBatchLoad = nonNull(loadResult, "Your batch loader function MUST return a non null CompletionStage promise");
        return mapBatchLoad.thenApply(map -> {
            List<V> values = new ArrayList<>();
            for (K key : keys) {
                V value = map.get(key);
                values.add(value);
            }
            return values;
        });
    }
494758
private void resumeTran(Transaction tran) {
        if (tran != null) {
            try {
                tranMgr.resume(tran);
            } catch (Exception e) {
                throw new BatchRuntimeException("Failed to resume transaction after JobOperator method", e);
            } 
        }
    }
345369
def _dequeue_batch(self) -> Optional[Batch]:
        
        if self._enqueueing_thread is None:
            raise ValueError('StreamWrapper `{}` with buffer of size `{}` was used outside with-resource environment.'
                             .format(self._name, self._buffer_size))
        if not self._enqueueing_thread.is_alive() and self._queue.empty():
            self._start_thread()
        while True:
            try:
                batch = self._queue.get(timeout=2)
                self._queue.task_done()
                break
            except Empty:
                if not self._enqueueing_thread.is_alive():
                    try:
                        # the enqueueing thread may just finished properly so lets check the queue eagerly
                        batch = self._queue.get_nowait()
                        self._queue.task_done()
                        break
                    except Empty:
                        # so we failed to retrieve a batch and the enqueueing thread is dead
                        # there is no hope, something must went wrong
                        raise ChildProcessError('Enqueueing thread ended unexpectedly.')
        return batch
1056012
private void doBatchWork(BatchBackend backend) throws InterruptedException {
		ExecutorService executor = Executors.newFixedThreadPool( typesToIndexInParallel, "BatchIndexingWorkspace" );
		for ( IndexedTypeIdentifier indexedTypeIdentifier : rootIndexedTypes ) {
			executor.execute( new BatchIndexingWorkspace( gridDialect, searchFactoryImplementor, sessionFactory, indexedTypeIdentifier,
					cacheMode, endAllSignal, monitor, backend, tenantId ) );
		}
		executor.shutdown();
		endAllSignal.await(); // waits for the executor to finish
	}
754278
function(callback){

    output("Initializing queries...");

    if(program.keyspace){
        batchQueries.unshift("USE " + program.keyspace + ";");
    }

    if(batchQueries && batchQueries.length > 0){
        output("Running queries...");
        executeQuery(batchQueries.shift(), callback);
    }else{
        callback("No queries to run.", null);
    }
}
786188
public void processEvent(QueryBatch batch) {
    if ( ! (batch.getClient() instanceof DatabaseClientImpl) ) {
      throw new IllegalStateException("DatabaseClient must be instanceof DatabaseClientImpl");
    }
    StringHandle uris = new StringHandle(String.join("\n", batch.getItems()))
      .withMimetype("text/uri-list");
    RESTServices services = ((DatabaseClientImpl) batch.getClient()).getServices();
    try {
      RequestParameters params = new RequestParameters();
      if ( transform != null ) transform.merge(params);
      params.add("result", applyResult.toString().toLowerCase());
      List<String> responseUris = new BufferedReader(
        services.postResource(null, "internal/apply-transform", null, params, uris, new ReaderHandle()).get()
      ).lines().collect(Collectors.toList());
      QueryBatchImpl processedBatch = new QueryBatchImpl()
        .withClient( batch.getClient() )
        .withItems( responseUris.toArray(new String[responseUris.size()]) )
        .withTimestamp( batch.getTimestamp() )
        .withJobBatchNumber( batch.getJobBatchNumber() )
        .withJobResultsSoFar( batch.getJobResultsSoFar() )
        .withForestBatchNumber( batch.getForestBatchNumber() )
        .withForestResultsSoFar( batch.getForestResultsSoFar() )
        .withForest( batch.getForest() )
        .withServerTimestamp( batch.getServerTimestamp() )
        .withJobTicket( batch.getJobTicket() );
      for ( QueryBatchListener listener : successListeners ) {
        try {
          listener.processEvent(processedBatch);
        } catch (Throwable t) {
          logger.error("Exception thrown by an onSuccess listener", t);
        }
      }

      List<String> skippedRequestUris = new ArrayList<>(Arrays.asList(batch.getItems()));
      skippedRequestUris.removeAll( responseUris );
      if ( skippedRequestUris.size() > 0 ) {
        QueryBatchImpl skippedBatch = processedBatch
          .withItems( skippedRequestUris.toArray(new String[0]) );
        for ( QueryBatchListener listener : skippedListeners ) {
          try {
            listener.processEvent(skippedBatch);
          } catch (Throwable t) {
            logger.error("Exception thrown by an onSkipped listener", t);
          }
        }
      }
    } catch (Throwable t) {
      for ( BatchFailureListener<Batch<String>> listener : failureListeners ) {
        try {
          listener.processFailure(batch, t);
        } catch (Throwable t2) {
          logger.error("Exception thrown by an onBatchFailure listener", t2);
        }
      }
      for ( BatchFailureListener<QueryBatch> queryBatchFailureListener : queryBatchFailureListeners ) {
        try {
          queryBatchFailureListener.processFailure(batch, t);
        } catch (Throwable t2) {
          logger.error("Exception thrown by an onFailure listener", t2);
        }
      }
      logger.warn("Error: {} in batch with urs ({})", t.toString(),
        Arrays.asList(batch.getItems()));
    }
  }
588849
@Override
  public void retryListener(QueryBatch batch, QueryBatchListener queryBatchListener) {
    // We get the batch and modify the client alone in order to make use
    // of the new forest client in case if the original host is unavailable.
    DatabaseClient client = null;
    Forest[] forests = batch.getBatcher().getForestConfig().listForests();
    for(Forest forest : forests) {
      if(forest.equals(batch.getForest()))
        client = getMoveMgr().getForestClient(forest);
    }
    QueryBatchImpl retryBatch = new QueryBatchImpl()
        .withClient( client )
        .withBatcher( batch.getBatcher() )
        .withTimestamp( batch.getTimestamp() )
        .withServerTimestamp( batch.getServerTimestamp() )
        .withItems( batch.getItems() )
        .withJobTicket( batch.getJobTicket() )
        .withJobBatchNumber( batch.getJobBatchNumber() )
        .withJobResultsSoFar( batch.getJobResultsSoFar() )
        .withForestBatchNumber( batch.getForestBatchNumber() )
        .withForestResultsSoFar( batch.getForestResultsSoFar() )
        .withForest( batch.getForest() )
        .withJobTicket( batch.getJobTicket() );
    queryBatchListener.processEvent(retryBatch);
  }
940343
void reportBatchFinished(ReceiveMessageBatchTask batch) {
        synchronized (finishedTasks) {
            finishedTasks.addLast(batch);
            if (log.isTraceEnabled()) {
                log.info("Queue " + qUrl + " now has " + finishedTasks.size() + " receive results cached ");
            }
        }
        synchronized (taskSpawnSyncPoint) {
            --inflightReceiveMessageBatches;
        }
        satisfyFuturesFromBuffer();
        spawnMoreReceiveTasks();
    }
656922
public void executeBatch(BatchOperation batchOperation) throws FMSException {

        IntuitMessage intuitMessage = prepareBatch(batchOperation);

        //execute interceptors
        executeInterceptors(intuitMessage);

        IntuitResponse intuitResponse = (IntuitResponse) intuitMessage.getResponseElements().getResponse();
        if (intuitResponse != null) {
            List<BatchItemResponse> batchItemResponses = intuitResponse.getBatchItemResponse();
            if (batchItemResponses != null && !batchItemResponses.isEmpty()) {
                int count = 0;
                Iterator<BatchItemResponse> itr = batchItemResponses.iterator();
                while (itr.hasNext()) {
                    BatchItemResponse batchItemResponse = itr.next();
                    String bId = batchItemResponse.getBId();
                    if (!StringUtils.hasText(bId)) {
                        bId = batchOperation.getBatchItemRequests().get(count).getBId();
                    }
                    if (batchItemResponse.getFault() != null) {
                        // segregate fault batch items
                        batchOperation.getFaultResult().put(bId, batchItemResponse.getFault());
                    } else if (batchItemResponse.getReport() != null) {
                        // segregate report batch items
                        batchOperation.getReportResult().put(bId, batchItemResponse.getReport());
                    } else if (batchItemResponse.getIntuitObject() != null) {
                        // segregate entity batch items
                        batchOperation.getEntityResult().put(bId, (IEntity) batchItemResponse.getIntuitObject().getValue());
                    } else if (batchItemResponse.getQueryResponse() != null) {
                        // segregate query batch items
                        QueryResult queryResult = getQueryResult(batchItemResponse.getQueryResponse());
                        batchOperation.getQueryResult().put(bId, queryResult);
                    } else if (batchItemResponse.getCDCResponse() != null) {
                        // segregate cdc query batch items
                        CDCQueryResult cdcQueryResult = getCDCQueryResult(batchItemResponse.getCDCResponse());
                        batchOperation.getCDCQueryResult().put(bId, cdcQueryResult);
                    } else {
                        LOG.warn("BatchItemResponse is not Fault, Entity, Query and Report.");
                    }
                    count++;
                }
            }
        }
    }
883902
private void execute(Map<Key, List<TableOperation>> batches)
    {
        if (batches != null && !batches.isEmpty())
        {
            try
            {
                for (List<TableOperation> batch : batches.values())
                {
                    tableAPI.execute(batch, null);
                }
            }
            catch (DurabilityException e)
            {
                log.error("Error while executing operations in OracleNOSQL, Caused by:" + e + ".");
                throw new PersistenceException("Error while Persisting data using batch", e);
            }
            catch (TableOpExecutionException e)
            {
                log.error("Error while executing operations in OracleNOSQL, Caused by:" + e + ".");
                throw new PersistenceException("Error while Persisting data using batch", e);
            }
            catch (FaultException e)
            {
                log.error("Error while executing operations in OracleNOSQL, Caused by:" + e + ".");
                throw new PersistenceException("Error while Persisting data using batch", e);
            }
            finally
            {
                batches.clear();
            }
        }
    }
458915
public synchronized void sendRetry(long retry, EventCallback callback) {

        if (open == 0 || shutdown) {
            if (callback != null) {
                callback.failed(this, null, null, null, new ClosedChannelException());
            }
            return;
        }
        queue.add(new SSEData(retry, callback));
        sink.getIoThread().execute(new Runnable() {
            @Override
            public void run() {
                synchronized (ServerSentEventConnection.this) {
                    if (pooled == null) {
                        fillBuffer();
                        writeListener.handleEvent(sink);
                    }
                }
            }
        });
    }
558559
@Override
    public IQueueMessage<ID, DATA> take() throws QueueException.EphemeralIsFull {
        if (!isEphemeralDisabled()) {
            int ephemeralMaxSize = getEphemeralMaxSize();
            if (ephemeralMaxSize > 0 && ephemeralSize() >= ephemeralMaxSize) {
                throw new QueueException.EphemeralIsFull(ephemeralMaxSize);
            }
        }
        lockTake.lock();
        try {
            if (lastFetchedId == null) {
                itQueue.seekToFirst();
            } else {
                itQueue.seek(lastFetchedId);
            }
            if (!itQueue.isValid()) {
                return null;
            }
            lastFetchedId = itQueue.key();
            byte[] value = itQueue.value();
            IQueueMessage<ID, DATA> msg = deserialize(value);
            try {
                batchTake.delete(cfQueue, lastFetchedId);
                // batchTake.put(cfMetadata, keyLastFetchedId, lastFetchedId);
                batchTake.put(cfMetadata, keyLastFetchedIdBytes, lastFetchedId);
                if (!isEphemeralDisabled() && msg != null) {
                    byte[] _key = msg.getId().toString().getBytes(QueueUtils.UTF8);
                    batchTake.put(cfEphemeral, _key, value);
                }
                rocksDbWrapper.write(writeOptions, batchTake);
            } catch (RocksDBException e) {
                throw new QueueException(e);
            } finally {
                batchTake.clear();
            }
            itQueue.next();
            return msg;
        } finally {
            lockTake.unlock();
        }
    }
1003935
@Override
    public int[] executeBatch() throws SQLException
    {
        checkClosed();
        closeCurrentResult();
        if (batch == null || batch.size() == 0) {
            return new int[0];
        }

        int[] updateCounts = new int[batch.size()];
        // keep a running total of update counts
        int runningUpdateCount = 0;

        int i = 0;
        try {
            for (; i < batch.size(); i++) {

                setCurrentResult(
                        null,
                        (int) batch.get(i).execute(
                                sourceConnection.NativeConnection,
                                this.m_timeout,
                                sourceConnection.queryTimeOutUnit)[0].fetchRow(
                                0).getLong(0));
                updateCounts[i] = this.lastUpdateCount;
                runningUpdateCount += this.lastUpdateCount;
            }
        } catch (SQLException x) {
            updateCounts[i] = EXECUTE_FAILED;
            throw new BatchUpdateException(Arrays.copyOf(updateCounts, i + 1),
                    x);
        } finally {
            clearBatch();
        }
        // replace the update count from the last statement with the update count
        // from the last batch.
        this.lastUpdateCount = runningUpdateCount;

        return updateCounts;
    }
273318
function reqeueBatch(times,interval,batch) {
    async.retry({times: times, interval: interval}, function(cb) {
      syncBatch(batch,cb);
    }, function done(err) {
      if (err) {
        log.error("failed to sync message batch", {err: err, data: batch});
      }
    });
  }
650982
private String recoverQueue(String queueName, QueueDeclaration queueDeclaration,
    List<Binding> queueBindings) throws Exception {
    String newQueueName = queueName;

    if (config.isQueueRecoveryEnabled()) {
      if (queueDeclaration != null) {
        newQueueName = recoverQueue(queueName, queueDeclaration);

        // Update dependencies for new queue names
        if (!queueName.equals(newQueueName)) {
          connectionHandler.queueDeclarations.remove(queueName);
          connectionHandler.queueDeclarations.put(newQueueName, queueDeclaration);
          connectionHandler.updateQueueBindingReferences(queueName, newQueueName);
        }
      }

      recoverQueueBindings(queueBindings);
    }

    return newQueueName;
  }