@Test
public void testRepairProcessingMessagesToFailed() throws Exception {

    setPrivateField(messageService, "countPartlyFailsBeforeFailed", new FixedConfigurationItem<>(2));
    RouteBuilder failedRoute = new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from(AsynchConstants.URI_ERROR_FATAL)
                .to("mock:test");
        }
    };
    getCamelContext().addRoutes(failedRoute);
    MockEndpoint failedMock = getCamelContext().getEndpoint("mock:test", MockEndpoint.class);
    failedMock.expectedMessageCount(1);
    Message message = createAndSaveMessages(1, new MessageCallback() {
        @Override
        public void beforeInsert(Message message, int order) {
            message.setState(MsgStateEnum.PROCESSING);
            message.setFailedCount(2);
            message.setMsgTimestamp(Instant.now().minus(1, ChronoUnit.HOURS));
            message.setStartProcessTimestamp(Instant.now().minus(1, ChronoUnit.HOURS));
            message.setLastUpdateTimestamp(Instant.now().minus(1, ChronoUnit.HOURS));
        }
    })[0];
    messageService.repairProcessingMessages();
    failedMock.assertIsSatisfied();
    Message found = messageDao.findMessage(message.getMsgId());
    assertThat(found, notNullValue());
    assertThat(found.getState(), is(MsgStateEnum.PROCESSING)); //"failed" route is mocked, state hasn't been changed
    // check
    Exchange exchange = failedMock.getExchanges().get(0);
    assertThat(exchange.getIn().getBody(), nullValue());
    assertThat(exchange.getProperty(Exchange.EXCEPTION_CAUGHT), notNullValue());
    assertThat(exchange.getProperty(Exchange.EXCEPTION_CAUGHT), instanceOf(IntegrationException.class));
    assertThat((Message)exchange.getIn().getHeader(AsynchConstants.MSG_HEADER), is(found));
}