@Override
  public ListenableFuture<Void> sendEdits(
      final long segmentTxId, final long firstTxnId,
      final int numTxns, final byte[] data) {
    try {
      reserveQueueSpace(data.length);
    } catch (LoggerTooFarBehindException e) {
      return Futures.immediateFailedFuture(e);
    }
    
    // When this batch is acked, we use its submission time in order
    // to calculate how far we are lagging.
    final long submitNanos = Time.monotonicNowNanos();
    
    ListenableFuture<Void> ret = null;
    try {
      ret = singleThreadExecutor.submit(() -> {
        throwIfOutOfSync();

        final long rpcSendTimeNanos = Time.monotonicNowNanos();
        try {
          getProxy().journal(createReqInfo(), segmentTxId, firstTxnId, numTxns, data);
        } catch (IOException e) {
          QuorumJournalManager.LOG.warn("Remote journal {} failed to write txns {}-{}."
                  + " Will try to write to this JN again after the next log roll.",
              IPCLoggerChannel.this, firstTxnId, (firstTxnId + numTxns - 1), e);
          synchronized (IPCLoggerChannel.this) {
            outOfSync = true;
          }
          throw e;
        } finally {
          final long nowNanos = Time.monotonicNowNanos();
          final long rpcTimeMicros = TimeUnit.MICROSECONDS.convert(
              (nowNanos - rpcSendTimeNanos), TimeUnit.NANOSECONDS);
          final long endToEndTimeMicros = TimeUnit.MICROSECONDS.convert(
              (nowNanos - submitNanos), TimeUnit.NANOSECONDS);
          metrics.addWriteEndToEndLatency(endToEndTimeMicros);
          metrics.addWriteRpcLatency(rpcTimeMicros);
          if (rpcTimeMicros / 1000 > WARN_JOURNAL_MILLIS_THRESHOLD) {
            QuorumJournalManager.LOG.warn(
                "Took {}ms to send a batch of {} edits ({} bytes) to remote journal {}.",
                rpcTimeMicros / 1000, numTxns, data.length, IPCLoggerChannel.this);
          }
        }
        synchronized (IPCLoggerChannel.this) {
          highestAckedTxId = firstTxnId + numTxns - 1;
          lastAckNanos = submitNanos;
        }
        return null;
      });
    } finally {
      if (ret == null) {
        // it didn't successfully get submitted,
        // so adjust the queue size back down.
        unreserveQueueSpace(data.length);
      } else {
        // It was submitted to the queue, so adjust the length
        // once the call completes, regardless of whether it
        // succeeds or fails.
        Futures.addCallback(ret, new FutureCallback<Void>() {
          @Override
          public void onFailure(Throwable t) {
            unreserveQueueSpace(data.length);
          }

          @Override
          public void onSuccess(Void t) {
            unreserveQueueSpace(data.length);
          }
        }, MoreExecutors.directExecutor());
      }
    }
    return ret;
  }
@Test
  public void testSimpleCall() throws Exception {
    ch.sendEdits(1, 1, 3, FAKE_DATA).get();
    Mockito.verify(mockProxy).journal(Mockito.<RequestInfo>any(),
        Mockito.eq(1L), Mockito.eq(1L),
        Mockito.eq(3), Mockito.same(FAKE_DATA));
  }