processBufferWatermark.decrementAndGet();
incomingMessages.mark();
final Timer.Context tcx = processTime.time();
Message msg = event.getMessage();
LOG.debug("Starting to process message <{}>.", msg.getId());
<BUG>try {
handleMessage(msg);</BUG>
} catch (Exception e) {
LOG.warn("Unable to process message <{}>: {}", msg.getId(), e);
} finally {

processBufferWatermark.decrementAndGet();
incomingMessages.mark();
final Timer.Context tcx = processTime.time();
Message msg = event.getMessage();
LOG.debug("Starting to process message <{}>.", msg.getId());
try {
LOG.debug("Finished processing message <{}>. Writing to output buffer.", msg.getId());
handleMessage(msg);
} catch (Exception e) {
LOG.warn("Unable to process message <{}>: {}", msg.getId(), e);
} finally {
