Flush all queued rows even if they are smaller than the batch size. This does not
guarantee that they will be reinserted if any of them fail. To make sure all rows
are either inserted or failed definitively, call shutdown().
495922
Future<?> flushAllTableQueues() throws InterruptedException {
        return m_es.submit(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                loadTable(buildTable(), m_table);
                return true;
            }
        });
    }
1217717
private void flushUpdateQueue() {
        LOG.debug("Run flushUpdateQueue.");

        final int size = queue.size();
        if (size <= 0) {
            return;
        }

        LOG.debug("{} events in flushUpdateQueue.", size);

        final Set<TargetPoll> events = Sets.newHashSetWithExpectedSize(queue.size());
        final int drained = queue.drainTo(events);

        if (drained <= 0) {
            return;
        }

        try {
            events.stream().collect(Collectors.groupingBy(TargetPoll::getTenant)).forEach((tenant, polls) -> {
                final TransactionCallback<Void> createTransaction = status -> updateLastTargetQueries(tenant, polls);
                tenantAware.runAsTenant(tenant,
                        () -> DeploymentHelper.runInNewTransaction(txManager, "flushUpdateQueue", createTransaction));
            });
        } catch (final RuntimeException ex) {
            LOG.error("Failed to persist UpdateQueue content.", ex);
            return;
        }

        LOG.debug("{} events persisted.", drained);
    }
684007
public function flush()
    {
        if ($this->is_done) {
            throw new RuntimeException('This batch insert is already done');
        }

        $count_rows = count($this->rows);

        if ($count_rows > 0) {
            $this->connection->execute($this->sql_foundation . implode(', ', $this->rows));

            $this->rows = [];
            $this->total += $count_rows;
        }
    }
836458
private void checkAndFlush() {
        try {
            int nowSize = memoryQueue.size();
            if (nowSize == 0) {
                return;
            }
            List<JobLogPo> batch = new ArrayList<JobLogPo>();
            for (int i = 0; i < nowSize; i++) {
                JobLogPo jobLogPo = memoryQueue.poll();
                batch.add(jobLogPo);

                if (batch.size() >= batchFlushSize) {
                    flush(batch);
                }
            }
            if (batch.size() > 0) {
                flush(batch);
            }

        } finally {
            flushing.compareAndSet(true, false);
        }
    }
568859
def flush(self):
        
        local_sender = self.sender
        if not local_sender:
            return
        while True:
            # get at most send_buffer_size items and send them
            data = []
            while len(data) < local_sender.send_buffer_size:
                item = self.get()
                if not item:
                    break
                data.append(item)
            if len(data) == 0:
                break
            local_sender.send(data)
3123
@Override
	protected void flush() {
		// The Kafka 0.8 producer doesn't support flushing, we wait here
		// until all pending records are confirmed
		synchronized (pendingRecordsLock) {
			while (pendingRecords > 0) {
				try {
					pendingRecordsLock.wait();
				} catch (InterruptedException e) {
					// this can be interrupted when the Task has been cancelled.
					// by throwing an exception, we ensure that this checkpoint doesn't get confirmed
					throw new RuntimeException("Flushing got interrupted while checkpointing", e);
				}
			}
		}
	}
606945
public synchronized void flushInsert()
    {
        if (!insertForUpdateList.isEmpty())
        {
            int sent = this.submitInsert(insertForUpdateList);
            if (sent < insertForUpdateList.size())
            {
                this.submitInsertTaskToQueue(this.createInsertTaskForObjects(insertForUpdateList.subList(sent, insertForUpdateList.size())), processedInserts);
            }
            this.insertForUpdateList = new MithraFastList(insertBatchSize);
        }

        if (!insertList.isEmpty())
        {
            int sent = this.submitInsert(insertList);
            if (sent < insertList.size())
            {
                this.submitInsertTaskToQueue(this.createInsertTaskForObjects(insertList.subList(sent, insertList.size())), processedInserts);
            }
            this.insertList = new MithraFastList(insertBatchSize);
        }
    }
1012307
def flush(self):
        """"""
        if self.shutdown:
            return
        self.flush_buffers(force=True)
        self.queue.put(FLUSH_MARKER)
        self.queue.join()
1110281
boolean flush() {
    if (!dirty)
      return true;

    acquireExclusiveLock();
    try {

      final long timer = OProfiler.getInstance().startChrono();

      // FORCE THE WRITE OF THE BUFFER
      for (int i = 0; i < FORCE_RETRY; ++i) {
        try {
          buffer.force();
          dirty = false;
          break;
        } catch (Exception e) {
          OLogManager.instance().debug(this,
              "Cannot write memory buffer to disk. Retrying (" + (i + 1) + "/" + FORCE_RETRY + ")...");
          OMemoryWatchDog.freeMemory(FORCE_DELAY);
        }
      }

      if (dirty)
        OLogManager.instance().debug(this, "Cannot commit memory buffer to disk after %d retries", FORCE_RETRY);
      else
        OProfiler.getInstance().updateCounter("system.file.mmap.pagesCommitted", 1);

      OProfiler.getInstance().stopChrono("system.file.mmap.commitPages", timer);

      return !dirty;

    } finally {
      releaseExclusiveLock();
    }
  }
724275
@Override
  public void flush() {
    try {
      out.flush();
    }
    catch(Exception ex) {
      reportError(null, ex, ErrorManager.FLUSH_FAILURE);
    }
    try {
      err.flush();
    }
    catch(Exception ex) {
      reportError(null, ex, ErrorManager.FLUSH_FAILURE);
    }
  }
14374
public function flush()
    {
        //todo bugfix batch record
        foreach ($this->deferLogs as $k => $deferLog) {
            $this->record($deferLog['message'], $deferLog['context'], $deferLog['level'], $deferLog['role']);
            $this->removeDeferLog($k);
        }
    }
694805
@Override
  public void flush()
  {
    // server/021g
    _logWriterQueue.wake();
    
    waitForFlush(10);
    
    try {
      super.flush();
    } catch (IOException e) {
      log.log(Level.FINER, e.toString(), e);
    }
  }
1240199
public void flush()
    {
        try {
            // Flush index writer
            indexWriter.flush();

            // Write out metrics mutations
            BatchWriter metricsWriter = connector.createBatchWriter(table.getMetricsTableName(), writerConfig);
            metricsWriter.addMutations(getMetricsMutations());
            metricsWriter.close();

            // Re-initialize the metrics
            metrics.clear();
            metrics.put(METRICS_TABLE_ROW_COUNT, new AtomicLong(0));
        }
        catch (MutationsRejectedException e) {
            throw new PrestoException(UNEXPECTED_ACCUMULO_ERROR, "Index mutation was rejected by server on flush", e);
        }
        catch (TableNotFoundException e) {
            throw new PrestoException(ACCUMULO_TABLE_DNE, "Accumulo table does not exist", e);
        }
    }
544627
public void flushTermination()
    {
        for (int i = 0; i < numberOfQueues; i++)
        {
            if (!terminateLists[i].isEmpty())
            {
                this.submitTaskToQueue(this.createTerminateTask(terminateLists[i]), i, processedTerminates);
                this.terminateLists[i] = new ArrayList(batchSize);
                this.terminatesQueued++;
            }
        }
    }
1009351
def flush(self, force=False):
    
    super(GCSRecordsPool, self).flush()
    if force:
      extra_padding = self._buf_size % self._GCS_BLOCK_SIZE
      if extra_padding > 0:
        self._write("\x00" * (self._GCS_BLOCK_SIZE - extra_padding))
    self._filehandle.flush()
581845
public function flush()
    {
        if ($this->recordSize === 0) {
            return;
        }

        $str = '';
        foreach ($this->records as $record) {
            $str .= $this->recordFormat($record);
        }

        $this->clear();
        $this->write($str);
    }
778615
def abort_incomplete_batches(self):
        
        # We need to keep aborting the incomplete batch until no thread is trying to append to
        # 1. Avoid losing batches.
        # 2. Free up memory in case appending threads are blocked on buffer full.
        # This is a tight loop but should be able to get through very quickly.
        while True:
            self._abort_batches()
            if not self._appends_in_progress.get():
                break
        # After this point, no thread will append any messages because they will see the close
        # flag set. We need to do the last abort after no thread was appending in case the there was a new
        # batch appended by the last appending thread.
        self._abort_batches()
        self._batches.clear()