leases now have a default duration of 60 seconds

Author: paullatzelsperger

extensions/transfer/transfer-store-cosmos/src/main/java/com/microsoft/dagx/transfer/store/cosmos/model/Lease.java

@@ -9,24 +9,23 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.time.Instant;
-import java.time.temporal.ChronoUnit;
 
 public class Lease {
     @JsonProperty
     private final String leasedBy;
     @JsonProperty
     private final long leasedAt;
     @JsonProperty
-    private final long leasedUntil;
+    private final long leaseDuration;
 
     Lease(String leasedBy) {
-        this(leasedBy, Instant.now().toEpochMilli(), Instant.now().plus(60, ChronoUnit.SECONDS).toEpochMilli());
+        this(leasedBy, Instant.now().toEpochMilli(), 60);
     }
 
-    public Lease(@JsonProperty("leasedBy") String leasedBy, @JsonProperty("leasedAt") long leasedAt, @JsonProperty("leasedUntil") long leasedUntil) {
+    public Lease(@JsonProperty("leasedBy") String leasedBy, @JsonProperty("leasedAt") long leasedAt, @JsonProperty("leaseDurationSeconds") long leaseDurationSeconds) {
         this.leasedBy = leasedBy;
         this.leasedAt = leasedAt;
-        this.leasedUntil = leasedUntil;
+        leaseDuration = leaseDurationSeconds;
     }
 
     public String getLeasedBy() {
@@ -37,7 +36,7 @@ public long getLeasedAt() {
         return leasedAt;
     }
 
-    public long getLeasedUntil() {
-        return leasedUntil;
+    public long getLeaseDuration() {
+        return leaseDuration;
     }
 }

extensions/transfer/transfer-store-cosmos/src/main/java/com/microsoft/dagx/transfer/store/cosmos/model/TransferProcessDocument.java

@@ -71,8 +71,8 @@ public void acquireLease(String connectorId) {
             lease = new Lease(connectorId);
         } else {
             var startDate = Instant.ofEpochMilli(lease.getLeasedAt());
-            var endDate = Instant.ofEpochMilli(lease.getLeasedUntil());
-            throw new IllegalStateException("This document is leased by " + lease.getLeasedBy() + "on " + startDate.toString() + " and cannot be leased again until " + endDate.toString() + "!");
+            var endDate = startDate.plusSeconds(lease.getLeaseDuration());
+            throw new IllegalStateException("This document is leased by " + lease.getLeasedBy() + "on " + startDate + " and cannot be leased again until " + endDate.toString() + "!");
         }
     }
 

extensions/transfer/transfer-store-cosmos/src/test/java/com/microsoft/dagx/transfer/store/cosmos/CosmosTransferProcessStoreTest.java

@@ -172,6 +172,7 @@ void nextForState_selfCanLeaseAgain() {
         var updatedDoc = readDocument(tp1.getId());
         assertThat(updatedDoc.getLease().getLeasedAt()).isNotEqualTo(originalTs);
         assertThat(doc.getLease().getLeasedBy()).isEqualTo(connectorId);
+        assertThat(doc.getLease().getLeaseDuration()).isEqualTo(60);
 
     }
 

extensions/transfer/transfer-store-cosmos/src/test/resources/lease.js

@@ -57,7 +57,8 @@ function lease(processId, connectorId, shouldLease) {
     function lease(document, connectorId) {
         document.lease = {
             leasedBy: connectorId,
-            leasedAt: Date.now()
+            leasedAt: Date.now(),
+            leaseDuration: 60
         };
 
         var accept = collection.replaceDocument(document._self, document, function (err, itemReplaced) {

extensions/transfer/transfer-store-cosmos/src/test/resources/nextForState.js

@@ -50,7 +50,8 @@ function nextForState(state, limit, connectorId) {
     function lease(document, connectorId) {
         document.lease = {
             leasedBy: connectorId,
-            leasedAt: Date.now()
+            leasedAt: Date.now(),
+            leaseDuration: 60
         };
 
         var accept = collection.replaceDocument(document._self, document, function (err, itemReplaced) {

scripts/lease.js

@@ -57,7 +57,8 @@ function lease(processId, connectorId, shouldLease) {
     function lease(document, connectorId) {
         document.lease = {
             leasedBy: connectorId,
-            leasedAt: Date.now()
+            leasedAt: Date.now(),
+            leaseDuration: 60
         };
 
         var accept = collection.replaceDocument(document._self, document, function (err, itemReplaced) {

scripts/nextForState.js

@@ -50,7 +50,8 @@ function nextForState(state, limit, connectorId) {
     function lease(document, connectorId) {
         document.lease = {
             leasedBy: connectorId,
-            leasedAt: Date.now()
+            leasedAt: Date.now(),
+            leaseDuration: 60
         };
 
         var accept = collection.replaceDocument(document._self, document, function (err, itemReplaced) {