@Override
    protected void doProcess(Exchange exchange, MetricsEndpoint endpoint, MetricRegistry registry, String metricsName)
            throws Exception {
        Message in = exchange.getIn();
        Meter meter = registry.meter(metricsName);
        Long mark = endpoint.getMark();
        Long finalMark = getLongHeader(in, HEADER_METER_MARK, mark);
        if (finalMark == null) {
            meter.mark();
        } else {
            meter.mark(finalMark);
        }
    }
@Test
    public void testProcessMarkSetOverrideByHeaderValue() throws Exception {
        when(endpoint.getMark()).thenReturn(MARK);
        when(in.getHeader(HEADER_METER_MARK, MARK, Long.class)).thenReturn(MARK + 101);
        producer.doProcess(exchange, endpoint, registry, METRICS_NAME);
        inOrder.verify(exchange, times(1)).getIn();
        inOrder.verify(registry, times(1)).meter(METRICS_NAME);
        inOrder.verify(endpoint, times(1)).getMark();
        inOrder.verify(in, times(1)).getHeader(HEADER_METER_MARK, MARK, Long.class);
        inOrder.verify(meter, times(1)).mark(MARK + 101);
        inOrder.verifyNoMoreInteractions();
    }