ErrorHandler { public boolean hasTooManyErrors(BaseTask.StreamMetrics metrics) { long msgsDone = metrics.processed.getCount() + metrics.dropped.getCount(); if (msgsDone > 100L) { double dropRate = metrics.dropped.getOneMinuteRate(); double successRate = metrics.processed.getOneMinuteRate(); if (dropRate > 0.0) { double totalRate = dropRate + successRate; double dropRatio = dropRate/totalRate; if (logger.isTraceEnabled()) { logger.trace(String.format("Drop rate: %f msg/s, success rate %f msg/s, total rate %f msg/s, drop ratio %f, max drop ratio %f", dropRate, successRate, totalRate, dropRatio, dropMaxRatio)); } if (dropRatio > dropMaxRatio) { logger.error(String.format("Error ratio (1min avg) %2f has exceeded threshold %f.", dropRatio, dropMaxRatio)); return true; } } else { logger.trace("Drop rate is still 0.0"); } } return false; } ErrorHandler(Config config, TaskInfo taskInfo); void start(); boolean dropOnError(); double getDropMaxRatio(); void handleException(IncomingMessageEnvelope envelope, Exception e, BaseTask.StreamMetrics metrics); boolean hasTooManyErrors(BaseTask.StreamMetrics metrics); void handleExpectedError(IncomingMessageEnvelope envelope, Exception e, BaseTask.StreamMetrics metrics); void stop();  }
@Test public void testNotTooManyErrors() throws Exception { ErrorHandler errHandler = getErrorHandler(); BaseTask.StreamMetrics metrics = new BaseTask.StreamMetrics(new Meter(clock), new Meter(clock)); when(clock.getTick()).thenReturn(0L); metrics.processed.mark(910L); metrics.dropped.mark(90L); when(clock.getTick()).thenReturn(TimeUnit.SECONDS.toNanos(10)); assertTrue(metrics.processed.getOneMinuteRate() > 0); assertTrue(metrics.dropped.getOneMinuteRate() > 0); assertFalse(errHandler.hasTooManyErrors(metrics)); }