Test fix for producer idempotence

Signed-off-by: Nithin <nithin.pankaj@walmart.com>
hyperledger-labs · May 30, 2024 · 08eb53d · 08eb53d
1 parent 486138b
commit 08eb53d
Show file tree

Hide file tree

Showing 7 changed files with 36 additions and 9 deletions.
diff --git a/src/main/java/hlf/java/rest/client/listener/ChaincodeEventListener.java b/src/main/java/hlf/java/rest/client/listener/ChaincodeEventListener.java
@@ -52,7 +52,7 @@ public void chaincodeEventListener(ContractEvent contractEvent) {
     }
 
     log.info(
-        "TxnID {} for Block Number {} is qualifies as a duplicate event.. Discarding the payload {} from being published.",
+        "TxnID {} for Block Number {} qualifies as a duplicate event.. Discarding the payload {} from being published.",
         txId,
         blockNumber,
         payload);

diff --git a/src/main/java/hlf/java/rest/client/service/impl/EventFulfillmentImpl.java b/src/main/java/hlf/java/rest/client/service/impl/EventFulfillmentImpl.java
@@ -92,13 +92,13 @@ public ResponseEntity<ClientResponseModel> replayEvents(
                     transactionActionInfo -> {
                       ChaincodeEvent chaincodeEvent = transactionActionInfo.getEvent();
 
-                      if (Objects.nonNull(transactionId)
-                          && chaincodeEvent.getTxId().equals(transactionId)) {
+                      if (Objects.isNull(transactionId)
+                          || chaincodeEvent.getTxId().equals(transactionId)) {
                         chaincodeEventListener.listener(
                             StringUtils.EMPTY, blockInfo, chaincodeEvent, networkName);
                       } else {
                         log.info(
-                            "Event TransactionID {} doesn't match with the provided TraansactionID filter {}, skipping event.",
+                            "Event TransactionID {} does not match the provided TransactionID filter {}. Skipping event.",
                             chaincodeEvent.getTxId(),
                             transactionId);
                       }

diff --git a/src/main/java/hlf/java/rest/client/service/impl/TransactionFulfillmentImpl.java b/src/main/java/hlf/java/rest/client/service/impl/TransactionFulfillmentImpl.java
@@ -223,6 +223,9 @@ public ResponseEntity<ClientResponseModel> writeTransactionToLedger(
 
       recencyTransactionContext.setTransactionContext(fabricTransaction.getTransactionId());
 
+      log.info(
+          "Performing Write Transaction to Ledger with Tx ID {}",
+          fabricTransaction.getTransactionId());
       byte[] result = fabricTransaction.submit(transactionParams);
       resultString = new String(result, StandardCharsets.UTF_8);
       log.info("Transaction Successfully Submitted - Response: " + resultString);
@@ -291,6 +294,9 @@ public ResponseEntity<ClientResponseModel> writePrivateTransactionToLedger(
 
       recencyTransactionContext.setTransactionContext(fabricTransaction.getTransactionId());
 
+      log.info(
+          "Performing Write Transaction to Ledger with Tx ID {}",
+          fabricTransaction.getTransactionId());
       byte[] result = fabricTransaction.submit(collection, transientKey);
       resultString = new String(result, StandardCharsets.UTF_8);
       log.info("Transaction Successfully Submitted - Response: " + resultString);
@@ -523,6 +529,9 @@ public ResponseEntity<ClientResponseModel> writeMultiDataTransactionToLedger(
 
       recencyTransactionContext.setTransactionContext(fabricTransaction.getTransactionId());
 
+      log.info(
+          "Performing Write Transaction to Ledger with Tx ID {}",
+          fabricTransaction.getTransactionId());
       byte[] result = fabricTransaction.submit(publicDataArgs);
 
       resultString = new String(result, StandardCharsets.UTF_8);

diff --git a/src/test/java/hlf/java/rest/client/service/impl/TransactionFulfillmentImplTest.java b/src/test/java/hlf/java/rest/client/service/impl/TransactionFulfillmentImplTest.java
@@ -8,6 +8,7 @@
 import hlf.java.rest.client.exception.FabricTransactionException;
 import hlf.java.rest.client.exception.ServiceException;
 import hlf.java.rest.client.model.ClientResponseModel;
+import hlf.java.rest.client.service.RecencyTransactionContext;
 import java.io.IOException;
 import java.util.List;
 import java.util.Optional;
@@ -42,6 +43,7 @@ public class TransactionFulfillmentImplTest {
   @Mock Network network;
   @Mock Contract contract;
   @Mock Transaction transaction;
+  @Mock RecencyTransactionContext recencyTransactionContext;
 
   private String testNetworkString = "some string";
   private String testContractString = "some string";
@@ -69,7 +71,9 @@ public void writeTransactionToLedgerTest()
     Mockito.when(gatewayConnetion.getNetwork(Mockito.anyString())).thenReturn(network);
     Mockito.when(network.getContract(Mockito.anyString())).thenReturn(contract);
     Mockito.when(contract.createTransaction(Mockito.anyString())).thenReturn(transaction);
+    Mockito.when(transaction.getTransactionId()).thenReturn("TX-1234");
     Mockito.when(transaction.submit(Mockito.<String>any())).thenReturn(byteArrayResponse);
+    Mockito.doNothing().when(recencyTransactionContext).setTransactionContext(Mockito.anyString());
     /*
      * Mockito.when(contract.submitTransaction(Mockito.anyString(),
      * Mockito.<String>any())) .thenReturn(byteArrayResponse);
@@ -92,7 +96,9 @@ public void writeTransactionToLedgerContractExceptionTest()
     Mockito.when(gatewayConnetion.getNetwork(Mockito.anyString())).thenReturn(network);
     Mockito.when(network.getContract(Mockito.anyString())).thenReturn(contract);
     Mockito.when(contract.createTransaction(Mockito.anyString())).thenReturn(transaction);
+    Mockito.when(transaction.getTransactionId()).thenReturn("TX-1234");
     Mockito.when(transaction.submit(Mockito.<String>any())).thenThrow(new ContractException(""));
+    Mockito.doNothing().when(recencyTransactionContext).setTransactionContext(Mockito.anyString());
     /*
      * Mockito.when(contract.submitTransaction(Mockito.anyString(),
      * Mockito.<String>any())) .thenThrow(new ContractException(""));
@@ -115,7 +121,9 @@ public void writeTransactionToLedgerTimeoutExceptionTest()
     Mockito.when(gatewayConnetion.getNetwork(Mockito.anyString())).thenReturn(network);
     Mockito.when(network.getContract(Mockito.anyString())).thenReturn(contract);
     Mockito.when(contract.createTransaction(Mockito.anyString())).thenReturn(transaction);
+    Mockito.when(transaction.getTransactionId()).thenReturn("TX-1234");
     Mockito.when(transaction.submit(Mockito.<String>any())).thenThrow(new TimeoutException());
+    Mockito.doNothing().when(recencyTransactionContext).setTransactionContext(Mockito.anyString());
     /*
      * Mockito.when(contract.submitTransaction(Mockito.anyString(),
      * Mockito.<String>any())) .thenThrow(new TimeoutException());
@@ -138,7 +146,9 @@ public void writeTransactionToLedgerInterruptExceptionTest()
     Mockito.when(gatewayConnetion.getNetwork(Mockito.anyString())).thenReturn(network);
     Mockito.when(network.getContract(Mockito.anyString())).thenReturn(contract);
     Mockito.when(contract.createTransaction(Mockito.anyString())).thenReturn(transaction);
+    Mockito.when(transaction.getTransactionId()).thenReturn("TX-1234");
     Mockito.when(transaction.submit(Mockito.<String>any())).thenThrow(new InterruptedException());
+    Mockito.doNothing().when(recencyTransactionContext).setTransactionContext(Mockito.anyString());
     /*
      * Mockito.when(contract.submitTransaction(Mockito.anyString(),
      * Mockito.<String>any())) .thenThrow(new InterruptedException());

diff --git a/src/test/resources/application-default-consumer.yml b/src/test/resources/application-default-consumer.yml
@@ -29,7 +29,8 @@ kafka:
   event-listener:
     ssl-enabled: false
     brokerHost: localhost:9092
-    topic: test-publisher-event-topic
+    topic: test-publisher-event-topice
+    enable-idempotence: true
 management:
   endpoints:
     web:
@@ -57,10 +58,12 @@ management:
         http.server.requests: 1s
     custom:
       kafka:
-        enabled: true
+        enabled: false
 server:
   config:
     location: /Users/n0p0328/work/hlf-connector/src/main/resources/config/
   servlet:
     context-path: /hlf-rest-client
   port: 8282
+dedupe:
+  enable: true
diff --git a/src/test/resources/application-per-partition-consumer.yml b/src/test/resources/application-per-partition-consumer.yml
@@ -31,10 +31,12 @@ kafka:
     ssl-enabled: false
     brokerHost: localhost:9092
     topic: test-publisher-event-topic
+    enable-idempotence: true
   failed-message-listener:
     brokerHost: localhost:9092
     topic: test-consumer-dlt
     ssl-enabled: false
+    enable-idempotence: true
 management:
   endpoints:
     web:
@@ -62,10 +64,12 @@ management:
         http.server.requests: 1s
     custom:
       kafka:
-        enabled: true
+        enabled: false
 server:
   config:
     location: /Users/n0p0328/work/hlf-connector/src/main/resources/config/
   servlet:
     context-path: /hlf-rest-client
   port: 8282
+dedupe:
+  enable: true
diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml
@@ -46,8 +46,9 @@ kafka:
   event-listener:
     brokerHost: localhost:9093
     topic: hlf-offchain-topic
-
-
+    enable-idempotence: true
+dedupe:
+  enable: false
 ---
 spring:
   profiles: intg