NIAD-3418: introducing retriable mechanism (#1329)

* introducing retriable mechanism

* fixing a failing test

* fixing flakiness of tests

* fixing flakiness of tests

* update changelog

---------

Co-authored-by: Jon Hardy <Jonathan.Hardy97@gmail.com>

Author: ORybak5

CHANGELOG.md

@@ -3,6 +3,9 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
+## [Unreleased]
+
+* Add Spring retry to MHS queue consumer to facilitate a more reasonable retry delay.
 
 ## [3.3.2] - 2026-08-19
 

gp2gp-translator/build.gradle

@@ -40,6 +40,8 @@ dependencies {
     implementation 'org.jdbi:jdbi3-core:3.49.5'
     implementation 'org.jetbrains:annotations:26.0.2'
     implementation 'org.springframework.cloud:spring-cloud-starter-gateway-server-webflux'
+    implementation 'org.springframework.retry:spring-retry:2.0.3'
+    implementation 'org.springframework.boot:spring-boot-starter-aop'
 
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation 'org.junit.jupiter:junit-jupiter:6.0.0'

gp2gp-translator/src/integrationTest/java/uk/nhs/adaptors/pss/translator/amqp/MhsQueueConsumerIT.java

@@ -0,0 +1,63 @@
+package uk.nhs.adaptors.pss.translator.amqp;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.UUID;
+
+import jakarta.jms.Message;
+import jakarta.jms.Session;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.bean.override.mockito.MockitoBean;
+import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
+
+import uk.nhs.adaptors.common.service.MDCService;
+import uk.nhs.adaptors.pss.translator.Gp2gpTranslatorApplication;
+import uk.nhs.adaptors.pss.translator.task.MhsQueueMessageHandler;
+
+@SpringBootTest(classes = Gp2gpTranslatorApplication.class)
+@TestPropertySource(properties = "amqp.daisyChaining=false")
+public class MhsQueueConsumerIT {
+
+    private static final String DELIVERY_COUNT_PROPERTY = "JMSXDeliveryCount";
+    public static final int RETRY_ATTEMPTS = 3;
+
+    @Autowired
+    private MhsQueueConsumer mhsQueueConsumer;
+
+    @MockitoSpyBean
+    private MhsQueueMessageHandler mhsQueueMessageHandler;
+
+    @MockitoBean
+    private MhsDlqPublisher mhsDlqPublisher;
+
+    @MockitoBean
+    private MDCService mdcService;
+
+    @Test
+    public void When_ReceiveThrowsRuntimeException_Expect_ReceiveMethodRetriedThreeTimes() throws Exception {
+        Message message = mock(Message.class);
+        Session session = mock(Session.class);
+
+        when(message.getJMSMessageID()).thenReturn(UUID.randomUUID().toString());
+        when(message.getIntProperty(DELIVERY_COUNT_PROPERTY)).thenReturn(1);
+        doThrow(new RuntimeException("Test failure"))
+                .when(mhsQueueMessageHandler).handleMessage(message);
+
+        assertThrows(RuntimeException.class, () -> mhsQueueConsumer.receive(message, session));
+
+        verify(mhsQueueMessageHandler, times(RETRY_ATTEMPTS)).handleMessage(message);
+        verifyNoInteractions(mhsDlqPublisher);
+        verify(mdcService, times(RETRY_ATTEMPTS)).resetAllMdcKeys();
+    }
+
+}

gp2gp-translator/src/integrationTest/java/uk/nhs/adaptors/pss/translator/amqp/ServiceFailureIT.java

@@ -8,6 +8,7 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -41,6 +42,7 @@
 import org.springframework.http.HttpHeaders;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.web.reactive.function.client.WebClientRequestException;
@@ -52,6 +54,7 @@
 import uk.nhs.adaptors.common.enums.MigrationStatus;
 import uk.nhs.adaptors.common.model.TransferRequestMessage;
 import uk.nhs.adaptors.pss.translator.Gp2gpTranslatorApplication;
+import uk.nhs.adaptors.pss.translator.config.MhsQueueProperties;
 import uk.nhs.adaptors.pss.translator.config.PssQueueProperties;
 import uk.nhs.adaptors.pss.translator.exception.MhsServerErrorException;
 import uk.nhs.adaptors.pss.translator.service.MhsClientService;
@@ -67,26 +70,29 @@
 @ExtendWith({SpringExtension.class, MockitoExtension.class})
 @DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
 @AutoConfigureMockMvc
+@TestPropertySource(properties = {
+    "amqp.pss.queueName=pssQueueServiceFailureIT",
+    "amqp.mhs.queueName=mhsQueueServiceFailureIT"
+})
 public class ServiceFailureIT extends BaseEhrHandler {
 
-    public static final int TEN_SECONDS = 10000;
     private static final String LOSING_ASID = "LOSING_ASID";
     private static final String WINNING_ASID = "WINNING_ASID";
     private static final String STUB_BODY = "test Body";
     private static final int THIRTY_SECONDS = 30000;
     private static final long TWO_MINUTES_LONG = 2L;
-    private static final int FIVE_WANTED_NUMBER_OF_INVOCATIONS = 5;
     public static final String JSON_LARGE_MESSAGE_SCENARIO_3_UK_06_JSON = "/json/LargeMessage/Scenario_3/uk06.json";
     public static final String JSON_LARGE_MESSAGE_SCENARIO_3_COPC_JSON = "/json/LargeMessage/Scenario_3/copc.json";
     public static final String JSON_LARGE_MESSAGE_EXPECTED_BUNDLE_SCENARIO_3_JSON = "/json/LargeMessage/expectedBundleScenario3.json";
     public static final int RECEIVE_TIMEOUT_LIMIT = 50;
-    public static final int TWENTY = 20;
     private String conversationId;
 
     @Autowired
     private ObjectMapper objectMapper;
     @Autowired
     private PssQueueProperties pssQueueProperties;
+    @Autowired
+    private MhsQueueProperties mhsQueueProperties;
 
     @Autowired
     @Qualifier("jmsTemplateMhsDLQ")
@@ -134,7 +140,8 @@ public void When_SendingInitialRequest_WithMhsOutboundServerError_Expect_Migrati
 
         sendRequestToPssQueue(conversationId, patientNhsNumber);
 
-        await().until(() -> hasMigrationStatus(EHR_EXTRACT_REQUEST_ERROR, conversationId));
+        await().atMost(Duration.ofMinutes(TWO_MINUTES_LONG))
+            .until(() -> hasMigrationStatus(EHR_EXTRACT_REQUEST_ERROR, conversationId));
 
         verify(mhsClientService, timeout(THIRTY_SECONDS).times(pssQueueProperties.getMaxRedeliveries() + 1)
         ).send(any());
@@ -146,15 +153,16 @@ public void When_SendingInitialRequest_WithMhsOutboundServerError_Expect_Migrati
     @Test
     public void When_ReceivingEhrExtract_WithMhsOutboundServerError_Expect_MigrationHasProcessingError() {
         doThrow(MhsServerErrorException.class)
-            .doNothing()
             .when(sendContinueRequestHandler)
             .prepareAndSendRequest(any());
 
         sendInboundMessageToQueue(JSON_LARGE_MESSAGE_SCENARIO_3_UK_06_JSON);
 
-        await().until(() -> hasMigrationStatus(EHR_GENERAL_PROCESSING_ERROR, getConversationId()));
+        await().atMost(Duration.ofMinutes(TWO_MINUTES_LONG))
+            .until(() -> hasMigrationStatus(EHR_GENERAL_PROCESSING_ERROR, getConversationId()));
 
-        verify(sendContinueRequestHandler, times(1)).prepareAndSendRequest(any());
+        verify(sendContinueRequestHandler, timeout(THIRTY_SECONDS).atLeast(MhsQueueConsumer.RETRY_ATTEMPTS))
+            .prepareAndSendRequest(any());
 
         assertThat(getCurrentMigrationStatus(getConversationId())).isEqualTo(EHR_GENERAL_PROCESSING_ERROR);
     }
@@ -189,6 +197,7 @@ public void When_ReceivingCOPC_WithMhsServerErrorException_Expect_MessageSentToD
         sendInboundMessageToQueue(JSON_LARGE_MESSAGE_SCENARIO_3_UK_06_JSON);
 
         await().until(this::hasContinueMessageBeenReceived);
+        clearInvocations(mhsClientService);
 
         doThrow(MhsServerErrorException.class).when(mhsClientService).send(any());
 
@@ -200,7 +209,7 @@ public void When_ReceivingCOPC_WithMhsServerErrorException_Expect_MessageSentToD
 
         assertNotNull(messageSentToDlq);
         assertEquals(copcMessageInJsonFormat, ((JmsTextMessage) messageSentToDlq).getText());
-        verify(mhsClientService, times(FIVE_WANTED_NUMBER_OF_INVOCATIONS)).send(any());
+        verify(mhsClientService, times((mhsQueueProperties.getMaxRedeliveries() + 1) * MhsQueueConsumer.RETRY_ATTEMPTS)).send(any());
     }
 
     @Test

gp2gp-translator/src/main/java/uk/nhs/adaptors/pss/translator/Gp2gpTranslatorApplication.java

@@ -2,7 +2,9 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.retry.annotation.EnableRetry;
 
+@EnableRetry
 @SpringBootApplication
 public class Gp2gpTranslatorApplication {
 

gp2gp-translator/src/main/java/uk/nhs/adaptors/pss/translator/amqp/MhsQueueConsumer.java

@@ -6,6 +6,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.jms.annotation.JmsListener;
+import org.springframework.retry.annotation.Retryable;
 import org.springframework.stereotype.Component;
 
 import lombok.RequiredArgsConstructor;
@@ -25,9 +26,11 @@ public class MhsQueueConsumer {
     private final MhsDlqPublisher mhsDlqPublisher;
     private final MhsQueueProperties mhsQueueProperties;
     private final MDCService mdcService;
+    public static final int RETRY_ATTEMPTS = 3;
 
     @JmsListener(destination = "${amqp.mhs.queueName}", containerFactory = "mhsQueueJmsListenerFactory")
     @SneakyThrows
+    @Retryable(maxAttempts = RETRY_ATTEMPTS)
     public void receive(Message message, Session session) {
         String messageId = message.getJMSMessageID();
         int deliveryCount = message.getIntProperty("JMSXDeliveryCount");