@Test public void testFinalRetry() throws IOException { final MessageListener messageListener = new MessageListener() { @Override public void onMessage(Message message, Context context) { throw new IllegalStateException(); } @Override public void onErrorMessage(Message message, Context context) { } }; ManagedMessageListener managedMessageListener = spy(new ManagedMessageListenerImpl(context, messageListener, managedInputQueue, 0)); RabbitMQMessageReceiver receiver = new RabbitMQMessageReceiver(rabbitMQMessagingProviderConfiguration, rabbitMQQueueConfiguration, deadLetterQueueConfigurations, inputQueueConfiguration, managedMessageListener, context, "consumer1"); Channel channel = Mockito.mock(Channel.class); AMQP.BasicProperties.Builder basicPropertiesBuilder = new AMQP.BasicProperties.Builder(); List<Object> xDeaths = new ArrayList<>(); Map<String, Object> xDeath = new HashMap<>(); xDeath.put("reason", "expired"); xDeath.put("count", 1L); xDeath.put("exchange", "test.dle"); xDeath.put("time", new Date(System.currentTimeMillis())); xDeath.put("routing-keys", Arrays.asList("test.dlq2.name")); xDeath.put("queue", "test.dlq2.name"); xDeaths.add(xDeath); xDeath = new HashMap<>(); xDeath.put("reason", "expired"); xDeath.put("count", 1L); xDeath.put("exchange", "test.dle"); xDeath.put("time", new Date(System.currentTimeMillis())); xDeath.put("routing-keys", Arrays.asList("test.dlq1.name")); xDeath.put("queue", "test.dlq1.name"); xDeaths.add(xDeath); Map<String, Object> headers = new HashMap<>(); headers.put("x-death", xDeaths); basicPropertiesBuilder.headers(headers); AMQP.BasicProperties basicProperties = basicPropertiesBuilder.build(); QueueingConsumer queueingConsumer = new QueueingConsumer(channel); queueingConsumer.handleDelivery( "consumer.tag", new Envelope(0, true, "test.exchange", "test.queue.name"), basicProperties, new byte[0]); BlobStoreAPI blobStoreAPI = Mockito.mock(BlobStoreAPI.class); receiver.handleDelivery( queueingConsumer, channel, blobStoreAPI, log, managedMessageListener, context, null, false, false, inputQueueConfiguration.getDeadLetterQueues(), deadLetterQueueConfigurations); verify(channel).basicNack(0, false, false); ArgumentCaptor<Message> messageArgumentCaptor = ArgumentCaptor.forClass(Message.class); verify(managedMessageListener).onErrorMessage(messageArgumentCaptor.capture(), eq(context)); Message message = messageArgumentCaptor.getValue(); Assert.assertEquals( "Expected failed message", "true", message.getMessageHeader(MessagingConstants.FAILED_HEADER)); Assert.assertNotNull("Expected error message", message.getMessageHeader(MessagingConstants.ERROR_HEADER)); }
void handleDelivery( QueueingConsumer queueingConsumer, Channel channel, BlobStoreAPI blobStoreAPI, Logger log, ManagedMessageListener listener, Context context, SerializationManager serializationManager, boolean shouldGzip, boolean shouldLogMessageContent, List<String> deadLetterQueues, Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations) { try { QueueingConsumer.Delivery delivery = null; RabbitMQMessage message = null; try { delivery = queueingConsumer.nextDelivery(); message = new RabbitMQMessage(serializationManager, delivery, blobStoreAPI, shouldGzip, shouldLogMessageContent); if (message.isFailed()) { listener.onErrorMessage(message, context); return; } listener.onMessage(message, context); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (NullPointerException e) { if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); message.setFailed(true); listener.onErrorMessage(message, context); } catch (Exception e) { if (message == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } List<RabbitMQMessage.XDeath> deaths = message.getXDeaths(); if (!delivery.getEnvelope().isRedeliver() && deaths.isEmpty()) { log.warn("Unable to process message. Message will be immediately re-queued.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); return; } if (deadLetterQueues.isEmpty()) { log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); return; } if (deaths.isEmpty()) { RabbitMQQueueConfiguration queue = deadLetterQueueConfigurations.get(deadLetterQueues.get(0)); String exchange = queue.getExchangeName(); String routingKey = queue.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } String lastDeadLetterQueue = deaths.get(0).getQueue(); Iterator<String> deadLetterQueuesIter = deadLetterQueues.iterator(); while (deadLetterQueuesIter.hasNext()) { RabbitMQQueueConfiguration conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); if (lastDeadLetterQueue.equals(conf.getQueueName())) { if (deadLetterQueuesIter.hasNext()) { conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); String exchange = conf.getExchangeName(); String routingKey = conf.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } } } log.warn( "Unable to process message and no retry dead letter queues are left. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); context.getHealthCheckManager().flagAsUnhealthy("Error processing message " + e.getMessage()); } } catch (IOException e) { log.error("Error processing RabbitMQ message.", e); } }
RabbitMQMessageReceiver extends QueueReceiverImpl<RabbitMQQueueConfiguration> { void handleDelivery( QueueingConsumer queueingConsumer, Channel channel, BlobStoreAPI blobStoreAPI, Logger log, ManagedMessageListener listener, Context context, SerializationManager serializationManager, boolean shouldGzip, boolean shouldLogMessageContent, List<String> deadLetterQueues, Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations) { try { QueueingConsumer.Delivery delivery = null; RabbitMQMessage message = null; try { delivery = queueingConsumer.nextDelivery(); message = new RabbitMQMessage(serializationManager, delivery, blobStoreAPI, shouldGzip, shouldLogMessageContent); if (message.isFailed()) { listener.onErrorMessage(message, context); return; } listener.onMessage(message, context); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (NullPointerException e) { if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); message.setFailed(true); listener.onErrorMessage(message, context); } catch (Exception e) { if (message == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } List<RabbitMQMessage.XDeath> deaths = message.getXDeaths(); if (!delivery.getEnvelope().isRedeliver() && deaths.isEmpty()) { log.warn("Unable to process message. Message will be immediately re-queued.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); return; } if (deadLetterQueues.isEmpty()) { log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); return; } if (deaths.isEmpty()) { RabbitMQQueueConfiguration queue = deadLetterQueueConfigurations.get(deadLetterQueues.get(0)); String exchange = queue.getExchangeName(); String routingKey = queue.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } String lastDeadLetterQueue = deaths.get(0).getQueue(); Iterator<String> deadLetterQueuesIter = deadLetterQueues.iterator(); while (deadLetterQueuesIter.hasNext()) { RabbitMQQueueConfiguration conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); if (lastDeadLetterQueue.equals(conf.getQueueName())) { if (deadLetterQueuesIter.hasNext()) { conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); String exchange = conf.getExchangeName(); String routingKey = conf.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } } } log.warn( "Unable to process message and no retry dead letter queues are left. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); context.getHealthCheckManager().flagAsUnhealthy("Error processing message " + e.getMessage()); } } catch (IOException e) { log.error("Error processing RabbitMQ message.", e); } } }
RabbitMQMessageReceiver extends QueueReceiverImpl<RabbitMQQueueConfiguration> { void handleDelivery( QueueingConsumer queueingConsumer, Channel channel, BlobStoreAPI blobStoreAPI, Logger log, ManagedMessageListener listener, Context context, SerializationManager serializationManager, boolean shouldGzip, boolean shouldLogMessageContent, List<String> deadLetterQueues, Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations) { try { QueueingConsumer.Delivery delivery = null; RabbitMQMessage message = null; try { delivery = queueingConsumer.nextDelivery(); message = new RabbitMQMessage(serializationManager, delivery, blobStoreAPI, shouldGzip, shouldLogMessageContent); if (message.isFailed()) { listener.onErrorMessage(message, context); return; } listener.onMessage(message, context); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (NullPointerException e) { if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); message.setFailed(true); listener.onErrorMessage(message, context); } catch (Exception e) { if (message == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } List<RabbitMQMessage.XDeath> deaths = message.getXDeaths(); if (!delivery.getEnvelope().isRedeliver() && deaths.isEmpty()) { log.warn("Unable to process message. Message will be immediately re-queued.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); return; } if (deadLetterQueues.isEmpty()) { log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); return; } if (deaths.isEmpty()) { RabbitMQQueueConfiguration queue = deadLetterQueueConfigurations.get(deadLetterQueues.get(0)); String exchange = queue.getExchangeName(); String routingKey = queue.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } String lastDeadLetterQueue = deaths.get(0).getQueue(); Iterator<String> deadLetterQueuesIter = deadLetterQueues.iterator(); while (deadLetterQueuesIter.hasNext()) { RabbitMQQueueConfiguration conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); if (lastDeadLetterQueue.equals(conf.getQueueName())) { if (deadLetterQueuesIter.hasNext()) { conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); String exchange = conf.getExchangeName(); String routingKey = conf.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } } } log.warn( "Unable to process message and no retry dead letter queues are left. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); context.getHealthCheckManager().flagAsUnhealthy("Error processing message " + e.getMessage()); } } catch (IOException e) { log.error("Error processing RabbitMQ message.", e); } } RabbitMQMessageReceiver(
            RabbitMQMessagingProviderConfiguration messagingProviderConfiguration,
            RabbitMQQueueConfiguration queueConfiguration,
            Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations,
            InputQueueConfiguration inputQueueConfiguration,
            ManagedMessageListener messageListener,
            Context context,
            String consumerName); }
RabbitMQMessageReceiver extends QueueReceiverImpl<RabbitMQQueueConfiguration> { void handleDelivery( QueueingConsumer queueingConsumer, Channel channel, BlobStoreAPI blobStoreAPI, Logger log, ManagedMessageListener listener, Context context, SerializationManager serializationManager, boolean shouldGzip, boolean shouldLogMessageContent, List<String> deadLetterQueues, Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations) { try { QueueingConsumer.Delivery delivery = null; RabbitMQMessage message = null; try { delivery = queueingConsumer.nextDelivery(); message = new RabbitMQMessage(serializationManager, delivery, blobStoreAPI, shouldGzip, shouldLogMessageContent); if (message.isFailed()) { listener.onErrorMessage(message, context); return; } listener.onMessage(message, context); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (NullPointerException e) { if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); message.setFailed(true); listener.onErrorMessage(message, context); } catch (Exception e) { if (message == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } List<RabbitMQMessage.XDeath> deaths = message.getXDeaths(); if (!delivery.getEnvelope().isRedeliver() && deaths.isEmpty()) { log.warn("Unable to process message. Message will be immediately re-queued.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); return; } if (deadLetterQueues.isEmpty()) { log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); return; } if (deaths.isEmpty()) { RabbitMQQueueConfiguration queue = deadLetterQueueConfigurations.get(deadLetterQueues.get(0)); String exchange = queue.getExchangeName(); String routingKey = queue.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } String lastDeadLetterQueue = deaths.get(0).getQueue(); Iterator<String> deadLetterQueuesIter = deadLetterQueues.iterator(); while (deadLetterQueuesIter.hasNext()) { RabbitMQQueueConfiguration conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); if (lastDeadLetterQueue.equals(conf.getQueueName())) { if (deadLetterQueuesIter.hasNext()) { conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); String exchange = conf.getExchangeName(); String routingKey = conf.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } } } log.warn( "Unable to process message and no retry dead letter queues are left. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); context.getHealthCheckManager().flagAsUnhealthy("Error processing message " + e.getMessage()); } } catch (IOException e) { log.error("Error processing RabbitMQ message.", e); } } RabbitMQMessageReceiver(
            RabbitMQMessagingProviderConfiguration messagingProviderConfiguration,
            RabbitMQQueueConfiguration queueConfiguration,
            Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations,
            InputQueueConfiguration inputQueueConfiguration,
            ManagedMessageListener messageListener,
            Context context,
            String consumerName); RabbitMQQueueConfiguration getQueueConfiguration(); @Override void init(); @Override void start(); @Override void pause(); @Override void cleanUp(); }
RabbitMQMessageReceiver extends QueueReceiverImpl<RabbitMQQueueConfiguration> { void handleDelivery( QueueingConsumer queueingConsumer, Channel channel, BlobStoreAPI blobStoreAPI, Logger log, ManagedMessageListener listener, Context context, SerializationManager serializationManager, boolean shouldGzip, boolean shouldLogMessageContent, List<String> deadLetterQueues, Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations) { try { QueueingConsumer.Delivery delivery = null; RabbitMQMessage message = null; try { delivery = queueingConsumer.nextDelivery(); message = new RabbitMQMessage(serializationManager, delivery, blobStoreAPI, shouldGzip, shouldLogMessageContent); if (message.isFailed()) { listener.onErrorMessage(message, context); return; } listener.onMessage(message, context); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (NullPointerException e) { if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); message.setFailed(true); listener.onErrorMessage(message, context); } catch (Exception e) { if (message == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } message.addErrorMessageHeader( context.getMicroServiceBaseURI(), System.currentTimeMillis(), 500, e.getMessage()); if (delivery == null) { log.warn("Unable to process message. Will wait for redelivery.", e); return; } List<RabbitMQMessage.XDeath> deaths = message.getXDeaths(); if (!delivery.getEnvelope().isRedeliver() && deaths.isEmpty()) { log.warn("Unable to process message. Message will be immediately re-queued.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); return; } if (deadLetterQueues.isEmpty()) { log.warn("Unable to process message. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); return; } if (deaths.isEmpty()) { RabbitMQQueueConfiguration queue = deadLetterQueueConfigurations.get(deadLetterQueues.get(0)); String exchange = queue.getExchangeName(); String routingKey = queue.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } String lastDeadLetterQueue = deaths.get(0).getQueue(); Iterator<String> deadLetterQueuesIter = deadLetterQueues.iterator(); while (deadLetterQueuesIter.hasNext()) { RabbitMQQueueConfiguration conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); if (lastDeadLetterQueue.equals(conf.getQueueName())) { if (deadLetterQueuesIter.hasNext()) { conf = deadLetterQueueConfigurations.get(deadLetterQueuesIter.next()); String exchange = conf.getExchangeName(); String routingKey = conf.getQueueName(); log.warn("Unable to process message. Message will be routed to dead letter queue '" + routingKey + "' bound to exchange '" + exchange + "'.", e); channel.basicPublish(exchange, routingKey, message.getProperties(), delivery.getBody()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); return; } } } log.warn( "Unable to process message and no retry dead letter queues are left. Message will be rejected.", e); channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); message.setFailed(true); listener.onErrorMessage(message, context); context.getHealthCheckManager().flagAsUnhealthy("Error processing message " + e.getMessage()); } } catch (IOException e) { log.error("Error processing RabbitMQ message.", e); } } RabbitMQMessageReceiver(
            RabbitMQMessagingProviderConfiguration messagingProviderConfiguration,
            RabbitMQQueueConfiguration queueConfiguration,
            Map<String, RabbitMQQueueConfiguration> deadLetterQueueConfigurations,
            InputQueueConfiguration inputQueueConfiguration,
            ManagedMessageListener messageListener,
            Context context,
            String consumerName); RabbitMQQueueConfiguration getQueueConfiguration(); @Override void init(); @Override void start(); @Override void pause(); @Override void cleanUp();  }