IPCLoggerChannel implements AsyncLogger { @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); } final long submitNanos = System.nanoTime(); ListenableFuture<Void> ret = null; try { ret = singleThreadExecutor.submit(new Callable<Void>() { @Override public Void call() throws IOException { throwIfOutOfSync(); long rpcSendTimeNanos = System.nanoTime(); try { getProxy().journal(createReqInfo(), segmentTxId, firstTxnId, numTxns, data); } catch (IOException e) { QuorumJournalManager.LOG.warn( "Remote journal " + IPCLoggerChannel.this + " failed to " + "write txns " + firstTxnId + "-" + (firstTxnId + numTxns - 1) + ". Will try to write to this JN again after the next " + "log roll.", e); synchronized (IPCLoggerChannel.this) { outOfSync = true; } throw e; } finally { long now = System.nanoTime(); long rpcTime = TimeUnit.MICROSECONDS.convert( now - rpcSendTimeNanos, TimeUnit.NANOSECONDS); long endToEndTime = TimeUnit.MICROSECONDS.convert( now - submitNanos, TimeUnit.NANOSECONDS); metrics.addWriteEndToEndLatency(endToEndTime); metrics.addWriteRpcLatency(rpcTime); if (rpcTime / 1000 > WARN_JOURNAL_MILLIS_THRESHOLD) { QuorumJournalManager.LOG.warn( "Took " + (rpcTime / 1000) + "ms to send a batch of " + numTxns + " edits (" + data.length + " bytes) to " + "remote journal " + IPCLoggerChannel.this); } } synchronized (IPCLoggerChannel.this) { highestAckedTxId = firstTxnId + numTxns - 1; lastAckNanos = submitNanos; } return null; } }); } finally { if (ret == null) { unreserveQueueSpace(data.length); } else { Futures.addCallback(ret, new FutureCallback<Void>() { @Override public void onFailure(Throwable t) { unreserveQueueSpace(data.length); } @Override public void onSuccess(Void t) { unreserveQueueSpace(data.length); } }); } } return ret; } IPCLoggerChannel(Configuration conf,
      NamespaceInfo nsInfo,
      String journalId,
      InetSocketAddress addr); @Override synchronized void setEpoch(long epoch); @Override synchronized void setCommittedTxId(long txid); @Override void close(); @Override URL buildURLToFetchLogs(long segmentTxId); synchronized int getQueuedEditsSize(); InetSocketAddress getRemoteAddress(); synchronized boolean isOutOfSync(); @Override ListenableFuture<Boolean> isFormatted(); @Override ListenableFuture<GetJournalStateResponseProto> getJournalState(); @Override ListenableFuture<NewEpochResponseProto> newEpoch(
      final long epoch); @Override ListenableFuture<Void> sendEdits(
      final long segmentTxId, final long firstTxnId,
      final int numTxns, final byte[] data); @Override ListenableFuture<Void> format(final NamespaceInfo nsInfo); @Override ListenableFuture<Void> startLogSegment(final long txid,
      final int layoutVersion); @Override ListenableFuture<Void> finalizeLogSegment(
      final long startTxId, final long endTxId); @Override ListenableFuture<Void> purgeLogsOlderThan(final long minTxIdToKeep); @Override ListenableFuture<RemoteEditLogManifest> getEditLogManifest(
      final long fromTxnId, final boolean inProgressOk); @Override ListenableFuture<PrepareRecoveryResponseProto> prepareRecovery(
      final long segmentTxId); @Override ListenableFuture<Void> acceptRecovery(
      final SegmentStateProto log, final URL url); @Override ListenableFuture<Void> discardSegments(final long startTxId); @Override ListenableFuture<Void> doPreUpgrade(); @Override ListenableFuture<Void> doUpgrade(final StorageInfo sInfo); @Override ListenableFuture<Void> doFinalize(); @Override ListenableFuture<Boolean> canRollBack(final StorageInfo storage,
      final StorageInfo prevStorage, final int targetLayoutVersion); @Override ListenableFuture<Void> doRollback(); @Override ListenableFuture<Long> getJournalCTime(); @Override String toString(); @Override synchronized void appendReport(StringBuilder sb); synchronized long getLagTxns(); synchronized long getLagTimeMillis();  }
@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)); }