When transaction ingestion failed return a failure response so the tr…

…ansaction composition consumer can handle it accordingly.
Backbase · Sep 8, 2023 · 533eaf8 · 533eaf8
1 parent fb247e8
commit 533eaf8
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 2 deletions.
diff --git a/...ckbase/stream/compositions/product/core/service/impl/ProductPostIngestionServiceImpl.java b/...ckbase/stream/compositions/product/core/service/impl/ProductPostIngestionServiceImpl.java
@@ -47,6 +47,7 @@ public Mono<ProductIngestResponse> handleSuccess(ProductIngestResponse res) {
                 .doOnNext(r -> log.info("Product ingestion completed successfully for SA {}",
                         res.getServiceAgreementInternalId()))
                 .flatMap(this::processTransactionChains)
+                .onErrorResume(x -> Mono.just(res))
                 .flatMap(this::processPaymentOrderChains)
                 .doOnNext(this::processSuccessEvent)
                 .doOnNext(r -> {

diff --git a/...tream/compositions/transaction/core/service/impl/TransactionPostIngestionServiceImpl.java b/...tream/compositions/transaction/core/service/impl/TransactionPostIngestionServiceImpl.java
@@ -47,6 +47,6 @@ public Mono<List<TransactionsPostResponseBody>> handleFailure(Throwable error) {
             envelopedEvent.setEvent(event);
             eventBus.emitEvent(envelopedEvent);
         }
-        return Mono.empty();
+        return Mono.error(error);
     }
 }
diff --git a/...tream/compositions/transaction/core/service/impl/TransactionIngestionServiceImplTest.java b/...tream/compositions/transaction/core/service/impl/TransactionIngestionServiceImplTest.java
@@ -164,7 +164,7 @@ void ingestionInPullMode_Failure() {
         Mono<TransactionIngestResponse> productIngestResponse = transactionIngestionService
                 .ingestPull(mockTransactionIngestPullRequest());
         StepVerifier.create(productIngestResponse)
-                .verifyComplete();
+                .verifyError();
     }
 
     @Test