Marmelator/redaction-service
1
0
Fork 0
Code Issues Pull Requests 14 Actions Packages Projects Releases 622 Wiki Activity

RED-5151: Do not retry messages on oom errors

Browse Source
This commit is contained in:
deiflaender 2022-09-26 12:52:39 +02:00
parent 142f5256ae
commit 0e71613ffc
4 changed files with 28 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
redaction-service-v1/redaction-service-server-v1/src/main/java/com/iqser/red/service/redaction/v1/server/queue/MessageReceiver.java
View File

@ -2,6 +2,7 @@ package com.iqser.red.service.redaction.v1.server.queue;
import static com.iqser.red.service.redaction.v1.server.queue.MessagingConfiguration.REDACTION_QUEUE; import static com.iqser.red.service.redaction.v1.server.queue.MessagingConfiguration.REDACTION_QUEUE;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListener;
@ -18,8 +19,8 @@ public class MessageReceiver {
@RabbitHandler @RabbitHandler
@RabbitListener(queues = REDACTION_QUEUE) @RabbitListener(queues = REDACTION_QUEUE)
public void receiveAnalyzeRequest(String in) throws JsonProcessingException { public void receiveAnalyzeRequest(Message message) {
redactionMessageReceiver.receiveAnalyzeRequest(in, false); redactionMessageReceiver.receiveAnalyzeRequest(message, false);
} }
} }

5
redaction-service-v1/redaction-service-server-v1/src/main/java/com/iqser/red/service/redaction/v1/server/queue/PriorityMessageReceiver.java
View File

@ -2,6 +2,7 @@ package com.iqser.red.service.redaction.v1.server.queue;
import static com.iqser.red.service.redaction.v1.server.queue.MessagingConfiguration.REDACTION_PRIORITY_QUEUE; import static com.iqser.red.service.redaction.v1.server.queue.MessagingConfiguration.REDACTION_PRIORITY_QUEUE;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListener;
@ -19,9 +20,9 @@ public class PriorityMessageReceiver {
@RabbitHandler @RabbitHandler
@RabbitListener(queues = REDACTION_PRIORITY_QUEUE) @RabbitListener(queues = REDACTION_PRIORITY_QUEUE)
public void receiveAnalyzeRequest(String in) throws JsonProcessingException { public void receiveAnalyzeRequest(Message message) {
redactionMessageReceiver.receiveAnalyzeRequest(in, true); redactionMessageReceiver.receiveAnalyzeRequest(message, true);
} }
} }

22
redaction-service-v1/redaction-service-server-v1/src/main/java/com/iqser/red/service/redaction/v1/server/queue/RedactionMessageReceiver.java
View File

@ -2,6 +2,8 @@ package com.iqser.red.service.redaction.v1.server.queue;
import static com.iqser.red.service.redaction.v1.server.queue.MessagingConfiguration.REDACTION_DQL; import static com.iqser.red.service.redaction.v1.server.queue.MessagingConfiguration.REDACTION_DQL;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@ -16,6 +18,7 @@ import com.iqser.red.service.redaction.v1.server.redaction.service.AnalyzeServic
import com.iqser.red.service.redaction.v1.server.redaction.service.ManualRedactionSurroundingTextService; import com.iqser.red.service.redaction.v1.server.redaction.service.ManualRedactionSurroundingTextService;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@Slf4j @Slf4j
@ -29,11 +32,24 @@ public class RedactionMessageReceiver {
private final ManualRedactionSurroundingTextService manualRedactionSurroundingTextService; private final ManualRedactionSurroundingTextService manualRedactionSurroundingTextService;
public void receiveAnalyzeRequest(String in, boolean priority) throws JsonProcessingException { @SneakyThrows
public void receiveAnalyzeRequest(Message message, boolean priority) {
var analyzeRequest = objectMapper.readValue(message.getBody(), AnalyzeRequest.class);
// This prevents from endless retries oom errors.
if(message.getMessageProperties().isRedelivered()){
throw new AmqpRejectAndDontRequeueException(String.format("Error during last processing of request with dossierId: %s and fileId: %s, do not retry.", analyzeRequest.getDossierId(), analyzeRequest.getFileId()));
}
receiveAnalyzeRequest(analyzeRequest, priority);
}
@SneakyThrows
public void receiveAnalyzeRequest(AnalyzeRequest analyzeRequest, boolean priority) {
var analyzeRequest = objectMapper.readValue(in, AnalyzeRequest.class);
log.info("Processing priority: {} analyze request for file: {}", priority, analyzeRequest.getFileId()); log.info("Processing priority: {} analyze request for file: {}", priority, analyzeRequest.getFileId());
AnalyzeResult result = null; AnalyzeResult result;
try { try {
switch (analyzeRequest.getMessageType()) { switch (analyzeRequest.getMessageType()) {

9
redaction-service-v1/redaction-service-server-v1/src/test/java/com/iqser/red/service/redaction/v1/server/realdata/AnalyseFileRealDataIntegrationTest.java
View File

@ -64,8 +64,7 @@ public class AnalyseFileRealDataIntegrationTest extends LiveDataIntegrationTest
log.info("No text file provided, Performing Structure analysis"); log.info("No text file provided, Performing Structure analysis");
ar.setMessageType(MessageType.STRUCTURE_ANALYSE); ar.setMessageType(MessageType.STRUCTURE_ANALYSE);
String in = om.writeValueAsString(ar); redactionMessageReceiver.receiveAnalyzeRequest(ar, false);
redactionMessageReceiver.receiveAnalyzeRequest(in, false);
} }
@ -75,15 +74,13 @@ public class AnalyseFileRealDataIntegrationTest extends LiveDataIntegrationTest
log.info("No redaction log provided, Performing full analysis"); log.info("No redaction log provided, Performing full analysis");
ar.setMessageType(MessageType.ANALYSE); ar.setMessageType(MessageType.ANALYSE);
String in = om.writeValueAsString(ar); redactionMessageReceiver.receiveAnalyzeRequest(ar, false);
redactionMessageReceiver.receiveAnalyzeRequest(in, false);
} }
simulateIncrement(List.of("Desiree"), "PII", 3L); simulateIncrement(List.of("Desiree"), "PII", 3L);
ar.setMessageType(MessageType.REANALYSE); ar.setMessageType(MessageType.REANALYSE);
String in = om.writeValueAsString(ar); redactionMessageReceiver.receiveAnalyzeRequest(ar, false);
redactionMessageReceiver.receiveAnalyzeRequest(in, false);
// var redactionLog = redactionStorageService.getRedactionLog("dossierId","fileId"); // var redactionLog = redactionStorageService.getRedactionLog("dossierId","fileId");