You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@omid.apache.org by fp...@apache.org on 2016/06/03 21:27:45 UTC
[18/23] incubator-omid git commit: [OMID-35] Fix and improve TSO
metrics
[OMID-35] Fix and improve TSO metrics
After refactoring the TSO pipeline the metrics, specially those related to
latencies, were not representative anymore. Also new specialized metrics,
have been added to measure special events.
Change-Id: Ie607872d0b2f3cf9e782eff70119e9d834979e43
Project: http://git-wip-us.apache.org/repos/asf/incubator-omid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-omid/commit/c4c65aca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-omid/tree/c4c65aca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-omid/diff/c4c65aca
Branch: refs/heads/master
Commit: c4c65acaef18651b047a1038d96fdd8ff367be69
Parents: 1f989f8
Author: Francisco Perez-Sorrosal <fp...@yahoo-inc.com>
Authored: Tue May 24 10:44:33 2016 -0700
Committer: Francisco Perez-Sorrosal <fp...@yahoo-inc.com>
Committed: Tue May 24 15:05:30 2016 -0700
----------------------------------------------------------------------
.../main/java/org/apache/omid/tso/Batch.java | 25 ++++++----
.../org/apache/omid/tso/MonitoringContext.java | 2 +-
.../omid/tso/PersistenceProcessorHandler.java | 48 ++++++++++++++------
.../org/apache/omid/tso/ReplyProcessorImpl.java | 46 ++++++++-----------
.../apache/omid/tso/RequestProcessorImpl.java | 25 +++++-----
.../org/apache/omid/tso/RetryProcessorImpl.java | 27 +++++++----
.../java/org/apache/omid/tso/TestBatch.java | 7 +++
.../org/apache/omid/tso/TestReplyProcessor.java | 26 ++++-------
8 files changed, 117 insertions(+), 89 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/main/java/org/apache/omid/tso/Batch.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/Batch.java b/tso-server/src/main/java/org/apache/omid/tso/Batch.java
index 496bca2..99d0c5c 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/Batch.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/Batch.java
@@ -93,36 +93,43 @@ public class Batch {
}
- void addCommit(long startTimestamp, long commitTimestamp, Channel c, MonitoringContext context) {
+ void addTimestamp(long startTimestamp, Channel c, MonitoringContext context) {
+
Preconditions.checkState(!isFull(), "batch is full");
int index = numEvents++;
PersistEvent e = events[index];
- e.makePersistCommit(startTimestamp, commitTimestamp, c, context);
+ context.timerStart("persistence.processor.timestamp.latency");
+ e.makePersistTimestamp(startTimestamp, c, context);
}
- public void addCommitRetry(long startTimestamp, Channel c, MonitoringContext context) {
+ void addCommit(long startTimestamp, long commitTimestamp, Channel c, MonitoringContext context) {
Preconditions.checkState(!isFull(), "batch is full");
int index = numEvents++;
PersistEvent e = events[index];
- e.makeCommitRetry(startTimestamp, c, context);
+ context.timerStart("persistence.processor.commit.latency");
+ e.makePersistCommit(startTimestamp, commitTimestamp, c, context);
}
- void addAbort(long startTimestamp, Channel c, MonitoringContext context) {
+ void addCommitRetry(long startTimestamp, Channel c, MonitoringContext context) {
+
Preconditions.checkState(!isFull(), "batch is full");
int index = numEvents++;
PersistEvent e = events[index];
- e.makePersistAbort(startTimestamp, c, context);
+ context.timerStart("persistence.processor.commit-retry.latency");
+ e.makeCommitRetry(startTimestamp, c, context);
}
- void addTimestamp(long startTimestamp, Channel c, MonitoringContext context) {
+ void addAbort(long startTimestamp, Channel c, MonitoringContext context) {
+
Preconditions.checkState(!isFull(), "batch is full");
int index = numEvents++;
PersistEvent e = events[index];
- e.makePersistTimestamp(startTimestamp, c, context);
+ context.timerStart("persistence.processor.abort.latency");
+ e.makePersistAbort(startTimestamp, c, context);
}
@@ -136,7 +143,7 @@ public class Batch {
.toString();
}
- public static class BatchFactory extends BasePooledObjectFactory<Batch> {
+ static class BatchFactory extends BasePooledObjectFactory<Batch> {
private static int batchId = 0;
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/main/java/org/apache/omid/tso/MonitoringContext.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/MonitoringContext.java b/tso-server/src/main/java/org/apache/omid/tso/MonitoringContext.java
index 4f448a5..426df27 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/MonitoringContext.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/MonitoringContext.java
@@ -37,7 +37,7 @@ public class MonitoringContext {
private static final Logger LOG = LoggerFactory.getLogger(MonitoringContext.class);
private volatile boolean flag;
- private Map<String, Long> elapsedTimeMsMap = new HashMap<>();
+ private Map<String, Long> elapsedTimeMsMap = new ConcurrentHashMap<>();
private Map<String, Stopwatch> timers = new ConcurrentHashMap<>();
private MetricsRegistry metrics;
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java b/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
index bfe0036..07241f0 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
@@ -79,21 +79,24 @@ public class PersistenceProcessorHandler implements WorkHandler<PersistenceProce
Batch batch = batchEvent.getBatch();
int numOfBatchedEvents = batch.getNumEvents();
batchSizeHistogram.update(numOfBatchedEvents);
- for (int i=0; i < numOfBatchedEvents; ++i) {
+ for (int i=0; i < numOfBatchedEvents; i++) {
PersistEvent event = batch.get(i);
-
switch (event.getType()) {
- case COMMIT:
- event.getMonCtx().timerStart("commitPersistProcessor");
- // TODO: What happens when the IOException is thrown?
- writer.addCommittedTransaction(event.getStartTimestamp(), event.getCommitTimestamp());
- commitEventsToFlush++;
- break;
- case ABORT:
- break;
- case TIMESTAMP:
- event.getMonCtx().timerStart("timestampPersistProcessor");
- break;
+ case TIMESTAMP:
+ event.getMonCtx().timerStop("persistence.processor.timestamp.latency");
+ break;
+ case COMMIT:
+ writer.addCommittedTransaction(event.getStartTimestamp(), event.getCommitTimestamp());
+ commitEventsToFlush++;
+ break;
+ case COMMIT_RETRY:
+ event.getMonCtx().timerStop("persistence.processor.commit-retry.latency");
+ break;
+ case ABORT:
+ event.getMonCtx().timerStop("persistence.processor.abort.latency");
+ break;
+ default:
+ throw new IllegalStateException("Event not allowed in Persistent Processor Handler: " + event);
}
}
@@ -101,6 +104,25 @@ public class PersistenceProcessorHandler implements WorkHandler<PersistenceProce
// to filter commit retries in the batch to disambiguate them.
flush(commitEventsToFlush);
filterAndDissambiguateClientRetries(batch);
+ for (int i=0; i < batch.getNumEvents(); i++) { // Just for statistics
+ PersistEvent event = batch.get(i);
+ switch (event.getType()) {
+ case TIMESTAMP:
+ event.getMonCtx().timerStart("reply.processor.timestamp.latency");
+ break;
+ case COMMIT:
+ event.getMonCtx().timerStop("persistence.processor.commit.latency");
+ event.getMonCtx().timerStart("reply.processor.commit.latency");
+ break;
+ case COMMIT_RETRY:
+ throw new IllegalStateException("COMMIT_RETRY events must be filtered before this step: " + event);
+ case ABORT:
+ event.getMonCtx().timerStart("reply.processor.abort.latency");
+ break;
+ default:
+ throw new IllegalStateException("Event not allowed in Persistent Processor Handler: " + event);
+ }
+ }
replyProcessor.manageResponsesBatch(batchEvent.getBatchSequence(), batch);
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/main/java/org/apache/omid/tso/ReplyProcessorImpl.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/ReplyProcessorImpl.java b/tso-server/src/main/java/org/apache/omid/tso/ReplyProcessorImpl.java
index a46c240..ab257d6 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/ReplyProcessorImpl.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/ReplyProcessorImpl.java
@@ -96,32 +96,30 @@ class ReplyProcessorImpl implements EventHandler<ReplyProcessorImpl.ReplyBatchEv
@VisibleForTesting
void handleReplyBatchEvent(ReplyBatchEvent replyBatchEvent) throws Exception {
- String name;
Batch batch = replyBatchEvent.getBatch();
for (int i = 0; i < batch.getNumEvents(); i++) {
PersistEvent event = batch.get(i);
switch (event.getType()) {
- case COMMIT:
- name = "commitReplyProcessor";
- event.getMonCtx().timerStart(name);
- sendCommitResponse(event.getStartTimestamp(), event.getCommitTimestamp(), event.getChannel());
- event.getMonCtx().timerStop(name);
- break;
- case ABORT:
- name = "abortReplyProcessor";
- event.getMonCtx().timerStart(name);
- sendAbortResponse(event.getStartTimestamp(), event.getChannel());
- event.getMonCtx().timerStop(name);
- break;
- case TIMESTAMP:
- name = "timestampReplyProcessor";
- event.getMonCtx().timerStart(name);
- sendTimestampResponse(event.getStartTimestamp(), event.getChannel());
- event.getMonCtx().timerStop(name);
- break;
- case COMMIT_RETRY:
- throw new RuntimeException("COMMIT_RETRY events must be filtered before this step. Event: {}");
+ case COMMIT:
+ sendCommitResponse(event.getStartTimestamp(), event.getCommitTimestamp(), event.getChannel());
+ event.getMonCtx().timerStop("reply.processor.commit.latency");
+ commitMeter.mark();
+ break;
+ case ABORT:
+ sendAbortResponse(event.getStartTimestamp(), event.getChannel());
+ event.getMonCtx().timerStop("reply.processor.abort.latency");
+ abortMeter.mark();
+ break;
+ case TIMESTAMP:
+ sendTimestampResponse(event.getStartTimestamp(), event.getChannel());
+ event.getMonCtx().timerStop("reply.processor.timestamp.latency");
+ timestampMeter.mark();
+ break;
+ case COMMIT_RETRY:
+ throw new IllegalStateException("COMMIT_RETRY events must be filtered before this step: " + event);
+ default:
+ throw new IllegalStateException("Event not allowed in Persistent Processor Handler: " + event);
}
event.getMonCtx().publish();
}
@@ -182,8 +180,6 @@ class ReplyProcessorImpl implements EventHandler<ReplyProcessorImpl.ReplyBatchEv
builder.setCommitResponse(commitBuilder.build());
c.write(builder.build());
- commitMeter.mark();
-
}
@Override
@@ -196,8 +192,6 @@ class ReplyProcessorImpl implements EventHandler<ReplyProcessorImpl.ReplyBatchEv
builder.setCommitResponse(commitBuilder.build());
c.write(builder.build());
- abortMeter.mark();
-
}
@Override
@@ -209,8 +203,6 @@ class ReplyProcessorImpl implements EventHandler<ReplyProcessorImpl.ReplyBatchEv
builder.setTimestampResponse(respBuilder.build());
c.write(builder.build());
- timestampMeter.mark();
-
}
final static class ReplyBatchEvent {
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/main/java/org/apache/omid/tso/RequestProcessorImpl.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/RequestProcessorImpl.java b/tso-server/src/main/java/org/apache/omid/tso/RequestProcessorImpl.java
index 32f5b8c..94a3df5 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/RequestProcessorImpl.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/RequestProcessorImpl.java
@@ -98,21 +98,15 @@ class RequestProcessorImpl implements EventHandler<RequestProcessorImpl.RequestE
@Override
public void onEvent(RequestEvent event, long sequence, boolean endOfBatch) throws Exception {
- String name = null;
- try { // TODO this should be a switch. Re-check why it's NOT now
- if (event.getType() == RequestEvent.Type.TIMESTAMP) {
- name = "timestampReqProcessor";
- event.getMonCtx().timerStart(name);
+ switch (event.getType()) {
+ case TIMESTAMP:
handleTimestamp(event);
- } else if (event.getType() == RequestEvent.Type.COMMIT) {
- name = "commitReqProcessor";
- event.getMonCtx().timerStart(name);
+ break;
+ case COMMIT:
handleCommit(event);
- }
- } finally {
- if (null != name) {
- event.getMonCtx().timerStop(name);
- }
+ break;
+ default:
+ throw new IllegalStateException("Event not allowed in Request Processor: " + event);
}
}
@@ -133,6 +127,7 @@ class RequestProcessorImpl implements EventHandler<RequestProcessorImpl.RequestE
@Override
public void timestampRequest(Channel c, MonitoringContext monCtx) {
+ monCtx.timerStart("request.processor.timestamp.latency");
long seq = requestRing.next();
RequestEvent e = requestRing.get(seq);
RequestEvent.makeTimestampRequest(e, c, monCtx);
@@ -144,6 +139,7 @@ class RequestProcessorImpl implements EventHandler<RequestProcessorImpl.RequestE
public void commitRequest(long startTimestamp, Collection<Long> writeSet, boolean isRetry, Channel c,
MonitoringContext monCtx) {
+ monCtx.timerStart("request.processor.commit.latency");
long seq = requestRing.next();
RequestEvent e = requestRing.get(seq);
RequestEvent.makeCommitRequest(e, startTimestamp, monCtx, writeSet, isRetry, c);
@@ -154,6 +150,7 @@ class RequestProcessorImpl implements EventHandler<RequestProcessorImpl.RequestE
private void handleTimestamp(RequestEvent requestEvent) throws Exception {
long timestamp = timestampOracle.next();
+ requestEvent.getMonCtx().timerStop("request.processor.timestamp.latency");
persistProc.addTimestampToBatch(timestamp, requestEvent.getChannel(), requestEvent.getMonCtx());
}
@@ -203,10 +200,12 @@ class RequestProcessorImpl implements EventHandler<RequestProcessorImpl.RequestE
persistProc.persistLowWatermark(newLowWatermark); // Async persist
}
}
+ event.getMonCtx().timerStop("request.processor.commit.latency");
persistProc.addCommitToBatch(startTimestamp, commitTimestamp, c, event.getMonCtx());
} else {
+ event.getMonCtx().timerStop("request.processor.commit.latency");
if (isCommitRetry) { // Re-check if it was already committed but the client retried due to a lag replying
persistProc.addCommitRetryToBatch(startTimestamp, c, event.getMonCtx());
} else {
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/main/java/org/apache/omid/tso/RetryProcessorImpl.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/RetryProcessorImpl.java b/tso-server/src/main/java/org/apache/omid/tso/RetryProcessorImpl.java
index 5ac12ad..a23a913 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/RetryProcessorImpl.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/RetryProcessorImpl.java
@@ -62,7 +62,9 @@ class RetryProcessorImpl implements EventHandler<RetryProcessorImpl.RetryEvent>,
final ObjectPool<Batch> batchPool;
// Metrics
- final Meter retriesMeter;
+ private final Meter txAlreadyCommittedMeter;
+ private final Meter invalidTxMeter;
+ private final Meter noCTFoundMeter;
@Inject
RetryProcessorImpl(MetricsRegistry metrics,
@@ -87,42 +89,47 @@ class RetryProcessorImpl implements EventHandler<RetryProcessorImpl.RetryEvent>,
retryExec.submit(retryProcessor);
// Metrics configuration
- retriesMeter = metrics.meter(name("tso", "retries"));
+ this.txAlreadyCommittedMeter = metrics.meter(name("tso", "retries", "commits", "tx-already-committed"));
+ this.invalidTxMeter = metrics.meter(name("tso", "retries", "aborts", "tx-invalid"));
+ this.noCTFoundMeter = metrics.meter(name("tso", "retries", "aborts", "tx-without-commit-timestamp"));
}
@Override
- public void onEvent(final RetryEvent event, final long sequence, final boolean endOfBatch)
- throws Exception {
+ public void onEvent(final RetryEvent event, final long sequence, final boolean endOfBatch) throws Exception {
switch (event.getType()) {
case COMMIT:
- // TODO: What happens when the IOException is thrown?
handleCommitRetry(event);
+ event.getMonCtx().timerStop("retry.processor.commit-retry.latency");
break;
default:
assert (false);
break;
}
+ event.getMonCtx().publish();
}
- private void handleCommitRetry(RetryEvent event) throws Exception {
+ private void handleCommitRetry(RetryEvent event) {
long startTimestamp = event.getStartTimestamp();
try {
Optional<CommitTimestamp> commitTimestamp = commitTableClient.getCommitTimestamp(startTimestamp).get();
if(commitTimestamp.isPresent()) {
if (commitTimestamp.get().isValid()) {
- LOG.trace("Valid commit TS found in Commit Table. Replying Commit to the client...");
+ LOG.trace("Tx {}: Valid commit TS found in Commit Table. Sending Commit to client.", startTimestamp);
replyProc.sendCommitResponse(startTimestamp, commitTimestamp.get().getValue(), event.getChannel());
+ txAlreadyCommittedMeter.mark();
} else {
- LOG.trace("Invalid commit TS found in Commit Table. Replying Abort to the client...");
+ LOG.trace("Tx {}: Invalid tx marker found. Sending Abort to client.", startTimestamp);
replyProc.sendAbortResponse(startTimestamp, event.getChannel());
+ invalidTxMeter.mark();
}
} else {
- LOG.trace("No commit TS found in Commit Table. Replying Abort to the client..");
+ LOG.trace("Tx {}: No Commit TS found in Commit Table. Sending Abort to client.", startTimestamp);
replyProc.sendAbortResponse(startTimestamp, event.getChannel());
+ noCTFoundMeter.mark();
}
} catch (InterruptedException e) {
LOG.error("Interrupted reading from commit table");
@@ -131,13 +138,13 @@ class RetryProcessorImpl implements EventHandler<RetryProcessorImpl.RetryEvent>,
LOG.error("Error reading from commit table", e);
}
- retriesMeter.mark();
}
@Override
public void disambiguateRetryRequestHeuristically(long startTimestamp, Channel c, MonitoringContext monCtx) {
long seq = retryRing.next();
RetryEvent e = retryRing.get(seq);
+ monCtx.timerStart("retry.processor.commit-retry.latency");
RetryEvent.makeCommitRetry(e, startTimestamp, c, monCtx);
retryRing.publish(seq);
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/test/java/org/apache/omid/tso/TestBatch.java
----------------------------------------------------------------------
diff --git a/tso-server/src/test/java/org/apache/omid/tso/TestBatch.java b/tso-server/src/test/java/org/apache/omid/tso/TestBatch.java
index 448eeaa..573cd89 100644
--- a/tso-server/src/test/java/org/apache/omid/tso/TestBatch.java
+++ b/tso-server/src/test/java/org/apache/omid/tso/TestBatch.java
@@ -22,8 +22,10 @@ import org.jboss.netty.channel.Channel;
import org.mockito.Mock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.mockito.MockitoAnnotations.initMocks;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
@@ -43,6 +45,11 @@ public class TestBatch {
@Mock
private MonitoringContext monCtx;
+ @BeforeMethod
+ void setup() {
+ initMocks(this);
+ }
+
@Test(timeOut = 10_000)
public void testBatchFunctionality() {
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/c4c65aca/tso-server/src/test/java/org/apache/omid/tso/TestReplyProcessor.java
----------------------------------------------------------------------
diff --git a/tso-server/src/test/java/org/apache/omid/tso/TestReplyProcessor.java b/tso-server/src/test/java/org/apache/omid/tso/TestReplyProcessor.java
index d7bebd6..3ae59df 100644
--- a/tso-server/src/test/java/org/apache/omid/tso/TestReplyProcessor.java
+++ b/tso-server/src/test/java/org/apache/omid/tso/TestReplyProcessor.java
@@ -20,12 +20,10 @@ package org.apache.omid.tso;
import org.apache.commons.pool2.ObjectPool;
import org.apache.omid.metrics.MetricsRegistry;
import org.apache.omid.metrics.NullMetricsProvider;
-import org.apache.omid.tso.PersistenceProcessorImpl.PersistBatchEvent;
import org.apache.omid.tso.ReplyProcessorImpl.ReplyBatchEvent;
import org.jboss.netty.channel.Channel;
import org.mockito.InOrder;
import org.mockito.Mock;
-import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,14 +31,8 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.io.IOException;
-
import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyInt;
-import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -48,7 +40,6 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertEqualsNoOrder;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -76,6 +67,9 @@ public class TestReplyProcessor {
@Mock
private Panicker panicker;
+ @Mock
+ private MonitoringContext monCtx;
+
private MetricsRegistry metrics;
private ObjectPool<Batch> batchPool;
@@ -112,7 +106,7 @@ public class TestReplyProcessor {
// Prepare test batch
Batch batch = batchPool.borrowObject();
- batch.addCommitRetry(FIRST_ST, null, null );
+ batch.addCommitRetry(FIRST_ST, null, monCtx);
ReplyBatchEvent e = ReplyBatchEvent.EVENT_FACTORY.newInstance();
ReplyBatchEvent.makeReplyBatch(e, batch, 0);
@@ -124,7 +118,7 @@ public class TestReplyProcessor {
try {
replyProcessor.onEvent(e, ANY_DISRUPTOR_SEQUENCE, false);
fail();
- } catch(RuntimeException re) {
+ } catch (RuntimeException re) {
// Expected
}
@@ -192,8 +186,8 @@ public class TestReplyProcessor {
// Prepare first a delayed batch (Batch #3)
Batch thirdBatch = batchPool.borrowObject();
- thirdBatch.addTimestamp(FIRST_ST, mock(Channel.class), mock(MonitoringContext.class));
- thirdBatch.addCommit(SECOND_ST, SECOND_CT, mock(Channel.class), mock(MonitoringContext.class));
+ thirdBatch.addTimestamp(FIRST_ST, mock(Channel.class), monCtx);
+ thirdBatch.addCommit(SECOND_ST, SECOND_CT, mock(Channel.class), monCtx);
ReplyBatchEvent thirdBatchEvent = ReplyBatchEvent.EVENT_FACTORY.newInstance();
ReplyBatchEvent.makeReplyBatch(thirdBatchEvent, thirdBatch, 2); // Set a higher sequence than the initial one
@@ -213,8 +207,8 @@ public class TestReplyProcessor {
// Prepare another delayed batch (Batch #2)
Batch secondBatch = batchPool.borrowObject();
- secondBatch.addTimestamp(THIRD_ST, mock(Channel.class), mock(MonitoringContext.class));
- secondBatch.addCommit(FOURTH_ST, FOURTH_CT, mock(Channel.class), mock(MonitoringContext.class));
+ secondBatch.addTimestamp(THIRD_ST, mock(Channel.class), monCtx);
+ secondBatch.addCommit(FOURTH_ST, FOURTH_CT, mock(Channel.class), monCtx);
ReplyBatchEvent secondBatchEvent = ReplyBatchEvent.EVENT_FACTORY.newInstance();
ReplyBatchEvent.makeReplyBatch(secondBatchEvent, secondBatch, 1); // Set another higher sequence
@@ -229,7 +223,7 @@ public class TestReplyProcessor {
// Finally, prepare the batch that should trigger the execution of the other two
Batch firstBatch = batchPool.borrowObject();
- firstBatch.addAbort(FIFTH_ST, mock(Channel.class), mock(MonitoringContext.class));
+ firstBatch.addAbort(FIFTH_ST, mock(Channel.class), monCtx);
ReplyBatchEvent firstBatchEvent = ReplyBatchEvent.EVENT_FACTORY.newInstance();
ReplyBatchEvent.makeReplyBatch(firstBatchEvent, firstBatch, 0); // Set the first batch with a higher sequence