class 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;
    }

}

class ErrorHandlerTest {

    private final Clock clock;

    @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));
    }

}