The buffers are bound by the number of tasks in them. Once the current
buffer has enough tasks, it will be queued and a new buffer will be
created.

@throws IOException
517237
private void bufferCatchup(int messageSize) throws IOException {
        // If the current buffer has too many tasks logged, queue it and
        // create a new one.
        if (m_tail != null && m_tail.size() > 0 && messageSize > m_bufferHeadroom) {
            // compile the invocation buffer
            m_tail.compile();

            final RejoinTaskBuffer boundTail = m_tail;
            final Runnable r = new Runnable() {
                @Override
                public void run() {
                    try {
                        m_buffers.offer(boundTail.getContainer());
                        if (m_reader.sizeInBytes() > m_overflowLimit * 1024 * 1024) {
                            // we can never catch up, should break rejoin.
                            VoltDB.crashLocalVoltDB("On-disk task log is full. Please reduce " +
                                    "workload and try live rejoin again, or use blocking rejoin.");
                        }
                    } catch (Throwable t) {
                        VoltDB.crashLocalVoltDB("Error in task log buffering transactions", true, t);
                    }
                }
            };

            m_es.execute(r);

            // Reset
            m_tail = null;
            m_tasksPendingInCurrentTail = 0;
        }

        // create a new buffer
        if (m_tail == null) {
            m_tail = new RejoinTaskBuffer(m_partitionId, messageSize);
            m_bufferHeadroom = RejoinTaskBuffer.DEFAULT_BUFFER_SIZE;
        }
    }
329333
private synchronized void initOut() throws IOException
   {
      if (out == null)
      {
         OutputStream os = new IndexOutputStream(dir.createOutput(REDO_LOG));
         out = new BufferedWriter(new OutputStreamWriter(os));
      }
   }
925663
public synchronized void enqueue(final byte[] array) throws IOException {
		assert array != null;
		byteDiskQueue.enqueueInt(array.length);
		byteDiskQueue.enqueue(array);
		size++;
	}
59692
protected void setupQueue() throws IOException {
		if (queueName != null) {
			channel.queueDeclare(queueName, false, false, false, null);
		}
	}
6716
private static void writeAndFlush(
      final ByteBuffer buffer, final OutputStream outputStream) throws IOException {

    if (buffer.hasArray()) {
      outputStream.write(buffer.array(), buffer.position(), buffer.remaining());
    } else {
      // cannot access underlying byte array, need to copy into a temporary array
      while (buffer.hasRemaining()) {
        // figure out how much to read, but use an upper limit of 8kb. LogMessages should be rather
        // small so we don't expect this to get hit but avoid large temporary buffers, just in case.
        final int size = Math.min(buffer.remaining(), 8 * 1024);
        final byte[] chunk = new byte[size];
        buffer.get(chunk);
        outputStream.write(chunk);
      }
    }
    outputStream.flush();
  }
883072
public void flushBuffer()
    throws IOException
    {
        if (logger.isLoggable(Level.FINER)) logger.entering(className, "flushBuffer()");

        if (this.osWriter != null)
        {
            this.osWriter.flush();
        }

        if (logger.isLoggable(Level.FINER)) logger.exiting(className, "flushBuffer()");
    }
1021630
void initNewBuffer(Writer out, int bufSize)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        { // 306998.15
            Tr.debug(tc, "initNewBuffer, size --> " + bufSize + " this --> " + this);
        }
        this.out = out;
        this.except = null;
        bufferSize = bufSize;
        buf = new char[bufferSize];
    }
1320546
protected synchronized void write(int b) throws IOException {
		if (closed) {
			throw new IOException("The buffer is closed");
		}

		verifyBufferSize(1);
		buf[writepos] = (byte) b;
		writepos++;

		notifyAll();
	}
724064
private void buffer() throws IOException {
        int remaining = buffer.length;
        while (remaining > 0) {
            int r = input.read(buffer, buffer.length - remaining, remaining);
            if (r <= 0) {
                // Not enough information to fill the buffer
                offset = 0;
                maxOffset = buffer.length - remaining;
                return;
            }
            remaining -= r;
        }
        maxOffset = buffer.length;
        offset = 0;
    }
650644
private void populateTasks(List<Flow> flows) throws IOException {
    if (flows == null || flows.size() == 0) {
      return;
    }

    // for simplicity, we assume that flows are ordered and consecutive
    JobKey startJob = null;
    // find the first job
    for (Flow f : flows) {
      List<JobDetails> jobs = f.getJobs();
      if (jobs != null && jobs.size() > 0) {
        startJob = jobs.get(0).getJobKey();
        break;
      }
    }

    if (startJob == null) {
      LOG.info("No start job found for flows");
      return;
    }

    byte[] startKey =
        Bytes.add(jobKeyConv.toBytes(startJob), Constants.SEP_BYTES);
    Scan scan = new Scan();
    scan.setStartRow(startKey);
    // expect a lot of tasks on average
    scan.setCaching(500);

    Table taskTable = hbaseConnection
        .getTable(TableName.valueOf(Constants.HISTORY_TASK_TABLE));
    ResultScanner scanner = taskTable.getScanner(scan);
    try {
      Result currentResult = scanner.next();
      for (Flow f : flows) {
        for (JobDetails j : f.getJobs()) {
          // within each job we advance through the scanner til we pass keys
          // matching the current job
          while (currentResult != null && !currentResult.isEmpty()) {
            TaskKey taskKey = taskKeyConv.fromBytes(currentResult.getRow());
            // see if this task belongs to the current job
            int comparison = j.getJobKey().compareTo(taskKey);
            if (comparison < 0) {
              // advance to next job (without advancing current result)
              break;
            } else if (comparison > 0) {
              // advance tasks up to current job
            } else {
              // belongs to the current job
              TaskDetails task = new TaskDetails(taskKey);
              task.populate(
                  currentResult.getFamilyMap(Constants.INFO_FAM_BYTES));
              j.addTask(task);
            }
            currentResult = scanner.next();
          }
          if (LOG.isDebugEnabled()) {
            LOG.debug("Added " + j.getTasks().size() + " tasks to job "
                + j.getJobKey().toString());
          }
        }
      }
    } finally {
      scanner.close();
      taskTable.close();
    }
  }
725719
@Override
	public void write(int b) throws IOException
    {
        if ((capacity - size) < 1)
            flush();
        buffer[size++] = (byte)b;
    }
197936
protected boolean checkBuffer() throws IOException {
        if (!enableMultiReadofPostData) {
            if (null != this.buffer) {
                if (this.buffer.hasRemaining()) {
                    return true;
                }
                this.buffer.release();
                this.buffer = null;
            }
            try {
                this.buffer = this.isc.getRequestBodyBuffer();
                if (null != this.buffer) {
                    // record the new amount of data read from the channel
                    this.bytesRead += this.buffer.remaining();
                    // Tr.debug(tc, "Buffer=" + WsByteBufferUtils.asString(this.buffer));
                    return true;
                }
                return false;
            } catch (IOException e) {
                this.error = e;
                throw e;
            }
        } else {
            return checkMultiReadBuffer();
        }
    }
1171711
@Override
    public void write(int value) throws IOException {
        validate();
        byte[] buf = new byte[1];
        buf[0] = (byte) value;
        writeToBuffers(buf, 0, 1);

    }
124145
public void seek(int offset) throws IOException {
    if (!(offset >= bufferOffset && offset < bufferOffset + maxBufferSize)) {
      writeBuffer();
      newBuffer(offset);
      aFile.seek(offset);
    }
    position = offset;
  }
981822
private void writeBuffer() throws IOException {
        bufferedWriter.flip(); // position is set to zero in dataFile.write
        dataFile.write(writeOffset, bufferedWriter);
        writeOffset += bufferedWriter.limit();
        bufferedWriter.clear();
        readNextChunkFromFile(); // before overwriting the next bucket. Read it.
    }
623405
public void commit() throws IOException {
        if(!marked)     throw new IllegalStateException();
        marked = false;
        super.append(buffer);
        buffer.setLength(0);
    }
561307
@Setup
  public void setup() throws IOException {
    byte[] source = new byte[8192];
    buffer = new Buffer();
    while (buffer.size() < bufferSize) {
      buffer.write(source);
    }
    cursor = new Buffer.UnsafeCursor();
  }