Merge pull request #936: [proxima-direct-core] O2-Czech-Republic#341 …

…fix DUPLICATE response handling
datadrivencz · Oct 22, 2024 · 2191bb0 · 2191bb0
2 parents c378f97 + 4bd2ca6
commit 2191bb0
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 0 deletions.
diff --git a/...c/main/java/cz/o2/proxima/direct/core/transaction/TransactionalOnlineAttributeWriter.java b/...c/main/java/cz/o2/proxima/direct/core/transaction/TransactionalOnlineAttributeWriter.java
@@ -199,6 +199,11 @@ public void update(List<KeyAttribute> addedInputs) {
         case OPEN:
           future = manager.updateTransaction(transactionId, addedInputs);
           break;
+        case COMMITTED:
+          future =
+              CompletableFuture.failedFuture(
+                  new TransactionRejectedException(transactionId, Flags.DUPLICATE));
+          break;
         default:
           future =
               CompletableFuture.failedFuture(

diff --git a/...st/java/cz/o2/proxima/direct/core/transaction/TransactionalOnlineAttributeWriterTest.java b/...st/java/cz/o2/proxima/direct/core/transaction/TransactionalOnlineAttributeWriterTest.java
@@ -407,6 +407,24 @@ public void testTransactionReopen() throws InterruptedException {
     assertEquals(Flags.DUPLICATE, res.take().getResponseFlags());
   }
 
+  @Test(timeout = 10000)
+  public void testTransactionReopenWithUpdate() throws InterruptedException {
+    TransactionalOnlineAttributeWriter writer = direct.getGlobalTransactionWriter();
+    assertTrue(user.isTransactional());
+    for (int i = 0; i < 2; i++) {
+      toReturn.add(Response.forRequest(anyRequest()).duplicate(1L));
+    }
+    String transactionId = UUID.randomUUID().toString();
+    BlockingQueue<TransactionRejectedException> res = new ArrayBlockingQueue<>(1);
+    try (Transaction t = writer.begin(transactionId)) {
+      t.beginGlobal();
+      t.update(List.of(KeyAttributes.ofMissingAttribute(gateway, "key", device, "dummy")));
+      t.commitWrite(
+          Collections.emptyList(), (succ, exc) -> res.add((TransactionRejectedException) exc));
+    }
+    assertEquals(Flags.DUPLICATE, res.take().getResponseFlags());
+  }
+
   @Test(timeout = 10_000)
   public void testWriteInTransactionWithTransform() throws InterruptedException {
     List<AttributeDescriptor<byte[]>> attrs = Arrays.asList(userGateways, gatewayUsers);