TransactionAttemptContext in Testcontainer behavior

Couchbase Server
1

Hi,

I am trying to test my code in couchbase testcontainer but facing certain anomaly.

My main code is following

public void execute() {
        while (!success) {
            try {
                transactions.run((TransactionAttemptContext ctx) -> {
                    process(ctx);
                });
                success = true;
            } catch (TransactionFailedException ex) {
}

And in my test code I am having:

when: "Stock receipt is received for that referenceId"
        String auditId = "test-1"
        def command = new StockReceivingProcessingCommand(storeStockMovement, stockProcessorHelper, srsProcessorHelper,
                auditId, missingPreReqHelper, transactions)
        command.execute()
        Thread.sleep(1000)

        then: "Expected transaction TransactionCode=#transactionCode ReasonCode=#reasonCode is created"
        if (transactionCode >= 0 && reasonCode >= 0) {
            QueryCriteria criteria = QueryCriteria.where("storeId").is(storeId)
                    .and(QueryCriteria.where("tpnb").is(rcvEventTpnb == null ? tpnb : rcvEventTpnb))
                    .and(QueryCriteria.where("referenceId").is(invoiceNo))
                    .and(QueryCriteria.where("transactionCode").is(transactionCode))
                    .and(QueryCriteria.where("reasonCode").is(reasonCode))

            def results = couchbaseTemplate.findByQuery(TransactionSubLedgerEntity)
                    .withConsistency(QueryScanConsistency.REQUEST_PLUS)
                    .matching(criteria)
                    .all()

Earlier when I was not having

transactions.run and calling process() directly my test case was able to fetch the data being saved in testcontainer.

But now its giving me empty while reading in then block, even though data is getting saved properly in the main code.
result is coming empty but not earlier.

Is there anything we need to handle for this.

2

Refer to the Spring Data Couchbase documentation for using transactions in Spring Data Couchbase.
https://docs.spring.io/spring-data/couchbase/reference/couchbase/transactions.html

Also refer to Couchbase Java SDK Transactions documentation for what Isolation is implemented. Developer Portal | Couchbase

btw - It’s not necessary to retry in a loop. The framework will retry exceptions that might succeed if retried.

3

Thanks for the reply.

My main issue is regarding the behavior inside the test container.

Where I am not able to see the data needed during assertion.
Here results is null.

Which was not the case when transaction.run was not happening.

4

If you follow the transaction documentation. you will get the correct behavior.

Just executing an operation inside a transaction lambda doesn’t mean the transaction is being used in the operation. The mechanism for executing using a transaction in spring-data-couchbase is via the spring-data-couchbase mechanism, not the Couchbase Java SDK transaction mechanism.

Non-transaction operations within a transaction lambda will not see transactional updates made within the same lambda

5

I am not able to give the entire code and context in one message.
My issue is with the TestContainer behavior
Like I am using the retry like this

    public void execute() {
        boolean success = false;
        int retryCount = 0;
        while (!success) {
            try {
                transactions.run((TransactionAttemptContext ctx) -> {
                    process(ctx);
                });
                success = true;
            } catch (TransactionFailedException ex) {
                Throwable cause = ex.getCause();

                if (cause instanceof DocumentExistsException || cause instanceof DuplicateKeyException) {
                    if (cause.getMessage().contains(getCollectionName(TransactionSubLedgerEntity.class))) {
                        log.warn("Duplicate event with eventId {} while stock-processing. Event already present",
                                storeStockMovement.getStockTransaction().stockTransactionId());
                        success = true;
                    } else {
                        retryCount++;
                        log.warn("Duplicate entry in stockWacSubLedger or stockRefSubLedger eventId {} while stock-processing. Retrying... Attempt {}",
                                storeStockMovement.getStockTransaction().stockTransactionId(), retryCount);

                        if (retryCount >= MAX_RETRY_COUNT) {
                            throw new TransactionProcessingException(
                                    String.format("Exceeded max retries while processing srs event with id %s",
                                            storeStockMovement.getStockTransaction().stockTransactionId()), ex);
                        }
                    }
                } else if (cause instanceof WacCalculationTransientException) {
                    throw (WacCalculationTransientException) cause;
                } else {
                    throw new TransactionProcessingException(
                            String.format("Error while processing stock event with id %s",
                                    storeStockMovement.getStockTransaction().stockTransactionId()), ex);
                }
            }
        }
    }

My own logic to retry.

Also, all db operations performed from the process(ctx) method uses ctx to maintain the transactional behavior.

For eg:

    public TransactionSubLedgerEntity saveEntity(TransactionAttemptContext ctx, TransactionSubLedgerEntity transactionSubLedgerEntity) {
        Collection collection = getCollection();

        try {
            ctx.insert(collection, transactionSubLedgerEntity.getTransactionId(), transactionSubLedgerEntity);
            log.info("Inserted new TransactionSubLedgerEntity with ID: {}", transactionSubLedgerEntity.getTransactionId());

        } catch (DocumentExistsException e) {
            log.error("Duplicate TransactionSubLedgerEntity detected with ID: {}", transactionSubLedgerEntity.getTransactionId());
            throw e;
        }
        return transactionSubLedgerEntity;
    }

But my main problem is when I am writing integration tests, where I use couchbase testcontainer

When I try to check the data I saved in my test code, its not coming up. Even though insertion has happened as per the logs in the code inside process(ctx) method.

Testing code piece

    def "Verify that successful stock receipt processing txn"() {
        given: "A SRS transfer is present in the system"
        buildDepotStockMovement(tpnb, storeId, referenceId, srsTransactionDate)
        insertStockWac(tpnb, storeId, 10)
        if(rcvEventTpnb!=tpnb)
            insertStockWac(rcvEventTpnb,storeId,11)
        insertStockRef(referenceId, 10, tpnb, storeId, stockInTransit, createdAt, srsTransactionDate)

        //currency  mock
        ConfigData configData = new ConfigData()
        configData.setCurrency("GBP")
        CurrencyMapping currencyMapping = new CurrencyMapping()
        def storeStockMovement =
                createStockReceivingEvent(rcvEventTpnb == null ? tpnb : rcvEventTpnb, storeId, invoiceNo, receiveQty, sourceTransactionDateTime)
        currencyMapping.setConfigData(Collections.singletonList(configData))
        Mockito.when(configurationService.getCurrencyMappingConfig(STOCK_CURRENCY_COUNTRY_MAPPING,
                storeStockMovement.getLocationReferenceEnriched().country()))
                .thenReturn(currencyMapping)
        Mockito.when(wacService.fetchCurrentWac(Mockito.anyString(), Mockito.any(LocalDate.class), Mockito.anyString()))
                .thenReturn(BigDecimal.valueOf(15.25))
        Mockito.when(missingPreReqHelper.persistMissingPreReqEvent()).thenReturn(new MissingPreReqEntity())

        when: "Stock receipt is received for that referenceId"
        String auditId = "test-1"
        def command = new StockReceivingProcessingCommand(storeStockMovement, stockProcessorHelper, srsProcessorHelper,
                auditId, missingPreReqHelper, transactions)
        command.execute()
        Thread.sleep(1000)

        then: "Expected transaction TransactionCode=#transactionCode ReasonCode=#reasonCode is created"
        if (transactionCode >= 0 && reasonCode >= 0) {
            QueryCriteria criteria = QueryCriteria.where("storeId").is(storeId)
                    .and(QueryCriteria.where("tpnb").is(rcvEventTpnb == null ? tpnb : rcvEventTpnb))
                    .and(QueryCriteria.where("referenceId").is(invoiceNo))
                    .and(QueryCriteria.where("transactionCode").is(transactionCode))
                    .and(QueryCriteria.where("reasonCode").is(reasonCode))

            def results = couchbaseTemplate.findByQuery(TransactionSubLedgerEntity)
                    .withConsistency(QueryScanConsistency.REQUEST_PLUS)
                    .matching(criteria)
                    .all()

            log.info("results {}", results);
            log.info("transaction {}", transactionSubLedgerRepository.findAll().toString())
            assert results.size() == 1
            assert results.stream().findFirst().get().quantity == expectedQty
            assert results.stream().findFirst().get().newStockOnHand == expectedSOH
        }
}

Here both the logs in results and transactions come as null.

6

ctx.insert(collection,

I don’t think your issue is transaction related.
Avoid mixing-and-matching the Couchbase Java SDK with Spring Data Couchbase. Spring Data Couchbase expects every document to have a _class property, and uses that as a predicate for all operations that are not -ById. So where you’ve used the Java SDK to insert a TransactionSubLedgerEntity document, it does not have a _class property. So when spring data queries with "SELECT … FROM … WHERE … AND _class = “com.example.TransactionSubLedgerEntity” it will only find documents inserted by Spring Data Couchbase (which have the _class property).
If you attempt to retrieve the document that you inserted using findById() - I believe you will find it.
There are exceptions, but by and large - that’s the deal.