code: public final Future<RecordMetadata> enqueue (D record, WriteCallback callback) throws InterruptedException {
      final ReentrantLock lock = this.dqLock;
      lock.lock();
      try {
        BytesBoundedBatch last = dq.peekLast();
        if (last != null) {
          Future<RecordMetadata> future = null;
          try {
            future = last.tryAppend(record, callback, this.largeMessagePolicy);
          } catch (RecordTooLargeException e) {
            // Ok if the record was too large for the current batch
          }
          if (future != null) {
            return future;
          }
        }
  
        // Create a new batch because previous one has no space
        BytesBoundedBatch batch = new BytesBoundedBatch(this.memSizeLimit, this.expireInMilliSecond);
        LOG.debug("Batch " + batch.getId() + " is generated");
        Future<RecordMetadata> future = null;
        try {
          future = batch.tryAppend(record, callback, this.largeMessagePolicy);
        } catch (RecordTooLargeException e) {
          // If a new batch also wasn't able to accomodate the new message
          throw new RuntimeException("Failed due to a message that was too large", e);
        }
  
        // The future might be null, since the largeMessagePolicy might be set to DROP
        if (future == null) {
          assert largeMessagePolicy.equals(LargeMessagePolicy.DROP);
          LOG.error("Batch " + batch.getId() + " is silently marked as complete, dropping a huge record: "
                  + record);
          future = Futures.immediateFuture(new RecordMetadata(0));
          callback.onSuccess(WriteResponse.EMPTY);
          return future;
        }
  
        // if queue is full, we should not add more
        while (dq.size() >= this.capacity) {
          LOG.debug("Accumulator size {} is greater than capacity {}, waiting", dq.size(), this.capacity);
          this.notFull.await();
        }
        dq.addLast(batch);
        incomplete.add(batch);
        this.notEmpty.signal();
        return future;
  
      } finally {
        lock.unlock();
      }
    }
docs: Add a data to internal deque data structure