MessagePollExecutor implements Runnable { @Override public void run() { LOG.debug("Message pooling starts ..."); Message msg = null; int lockFailureCount = 0; while (true) { try { msg = messagesPool.getNextMessage(); if (msg != null) { LogContextHelper.setLogContextParams(msg, null); startMessageProcessing(msg); } else { break; } } catch (LockFailureException ex) { lockFailureCount++; if (lockFailureCount > LOCK_FAILURE_LIMIT) { LOG.warn("Probably problem with locking messages - count of lock failures exceeds limit (" + LOCK_FAILURE_LIMIT + ")."); break; } } catch (Exception ex) { LOG.error("Error occurred during getting message " + (msg != null ? msg.toHumanString() : ""), ex); } } LOG.debug("Message pooling finished."); }  @Override void run();  }
@Test public void testGetNextMessage_moreThreads() throws InterruptedException { int threads = 5; final CountDownLatch latch = new CountDownLatch(threads); Runnable task = new Runnable() { @Override public void run() { try { messagePollExecutor.run(); } finally { latch.countDown(); } } }; mock.expectedMessageCount(3); for (int i = 0; i < threads; i++) { new Thread(task).start(); } latch.await(); mock.assertIsSatisfied(); Message msg = findMessage("1234_4567"); assertThat(msg, notNullValue()); assertThat(msg.getState(), is(MsgStateEnum.IN_QUEUE)); msg = findMessage("1234_4567_8"); assertThat(msg, notNullValue()); assertThat(msg.getState(), is(MsgStateEnum.IN_QUEUE)); msg = findMessage("1234_4567_9"); assertThat(msg, notNullValue()); assertThat(msg.getState(), is(MsgStateEnum.IN_QUEUE)); }