Я столкнулся с проблемой возможного исключения idempotencyBreachException в службе, используемой прослушивателем Kafka.
Я не совсем уверен, как правильно справиться с этой ситуацией. У меня есть lockExecutor, который блокирует заданный идентификатор клиента и кеш, используемый в качестве идемпотентного репозитория. В 99,99% случаев все будет работать так, как ожидалось, но меня беспокоит сценарий, когда в конечном итоге возникает исключение IdempotencyBreachException и в базе данных сохраняются 2xdata. Должен ли я просто зарегистрировать ошибку и обработать ее вручную, когда она произойдет, или есть какой-то лучший способ справиться с ней?
Я рассматривал возможность использования @Retryable в методе обработки, но это создает проблему: ключ кэша уже использован, и я знаю, что не следует его удалять. Кроме того, меня беспокоит, как в этом контексте повторные попытки будут взаимодействовать с откатами, поскольку это будет параллельная обработка. На данный момент единственное решение, которое приходит на ум, — это зарегистрировать исключение и при необходимости обработать его вручную в будущем.
@Service
@RequiredArgsConstructor
public class ProcessingEventsServiceImpl {
private final IdempotencyRepository idempotencyRepository;
private final TransactionalService transactionalService;
private final LockExecutor lockExecutor;
public void process(InboundData inbound) {
final String key = inbound.getTransactionId();
final String customerId = inbound.getCustomerId();
final Provenance provenance = inbound.getDataProvenance();
lockExecutor.run(customerId, () -> {
if (idempotencyRepository.isKeyUsed(provenance, key)) {
return;
}
transactionalService.executeTransactionProcess(inbound);
try {
idempotencyRepository.markKeyAsUsed(provenance, key);
} catch (IdempotencyBreachException e) {
log.error(...);
}
});
}
}
@Component
@RequiredArgsConstructor
public class KafkaListener {
private final ProcessingService processingService;
private final DataMapper mapper;
private static final Logger LOGGER =
LoggerFactory.getLogger(KafkaListener.class);
@KafkaListener(topics = "topic")
public void onEvent(Event event) {
var data = mapper.mapEventToModel(event);
processingService.process(data);
}
}
@Service
@RequiredArgsConstructor
public class TransactionalService {
private final InboundRepository repository;
private final InboundMapper mapper;
@Transactional
public void executeTransactionalProcess(TransferData inbound) {
var entity = repository.saveOrUpdate(inbound);
repository.save(entity);
}
}
Подробнее здесь: https://stackoverflow.com/questions/798 ... a-listener
