BUG-8792: allow transactions to not time out after reconnect

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / Shard.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
index d9483d7b2b61915295eef094fc8c8570f0b63eaa..715358b514555dfd7502475b128056d434ff4161 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
@@ -36,6 +36,7 @@ import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
 import org.opendaylight.controller.cluster.access.commands.LocalHistoryRequest;
 import org.opendaylight.controller.cluster.access.commands.NotLeaderException;
+import org.opendaylight.controller.cluster.access.commands.OutOfSequenceEnvelopeException;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
@@ -328,8 +329,7 @@ public class Shard extends RaftActor {
                 PeerAddressResolved resolved = (PeerAddressResolved) message;
                 setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
             } else if (TX_COMMIT_TIMEOUT_CHECK_MESSAGE.equals(message)) {
-                store.checkForExpiredTransactions(transactionCommitTimeout);
-                commitCoordinator.checkForExpiredTransactions(transactionCommitTimeout, this);
+                commitTimeoutCheck();
             } else if (message instanceof DatastoreContext) {
                 onDatastoreContext((DatastoreContext)message);
             } else if (message instanceof RegisterRoleChangeListener) {
@@ -390,6 +390,18 @@ public class Shard extends RaftActor {
         }
     }
 
+    private void commitTimeoutCheck() {
+        store.checkForExpiredTransactions(transactionCommitTimeout, this::updateAccess);
+        commitCoordinator.checkForExpiredTransactions(transactionCommitTimeout, this);
+    }
+
+    private Optional<Long> updateAccess(final SimpleShardDataTreeCohort cohort) {
+        final FrontendIdentifier frontend = cohort.getIdentifier().getHistoryId().getClientId().getFrontendId();
+        // If this frontend has freshly connected, give it some time to catch up before killing its transactions.
+        final LeaderFrontendState state = knownFrontends.get(frontend);
+        return state == null ? Optional.absent() : Optional.of(state.getLastConnectTicks());
+    }
+
     private void onMakeLeaderLocal() {
         LOG.debug("{}: onMakeLeaderLocal received", persistenceId());
         if (isLeader()) {
@@ -417,11 +429,13 @@ public class Shard extends RaftActor {
     }
 
     // Acquire our frontend tracking handle and verify generation matches
-    private LeaderFrontendState getFrontend(final ClientIdentifier clientId) throws RequestException {
+    @Nullable
+    private LeaderFrontendState findFrontend(final ClientIdentifier clientId) throws RequestException {
         final LeaderFrontendState existing = knownFrontends.get(clientId.getFrontendId());
         if (existing != null) {
             final int cmp = Long.compareUnsigned(existing.getIdentifier().getGeneration(), clientId.getGeneration());
             if (cmp == 0) {
+                existing.touch();
                 return existing;
             }
             if (cmp > 0) {
@@ -436,10 +450,17 @@ public class Shard extends RaftActor {
             LOG.debug("{}: client {} is not yet known", persistenceId(), clientId);
         }
 
-        final LeaderFrontendState ret = new LeaderFrontendState(persistenceId(), clientId, store);
-        knownFrontends.put(clientId.getFrontendId(), ret);
-        LOG.debug("{}: created state {} for client {}", persistenceId(), ret, clientId);
-        return ret;
+        return null;
+    }
+
+    private LeaderFrontendState getFrontend(final ClientIdentifier clientId) throws RequestException {
+        final LeaderFrontendState ret = findFrontend(clientId);
+        if (ret != null) {
+            return ret;
+        }
+
+        // TODO: a dedicated exception would be better, but this is technically true, too
+        throw new OutOfSequenceEnvelopeException(0);
     }
 
     private static @Nonnull ABIVersion selectVersion(final ConnectClientRequest message) {
@@ -458,6 +479,12 @@ public class Shard extends RaftActor {
     @SuppressWarnings("checkstyle:IllegalCatch")
     private void handleConnectClient(final ConnectClientRequest message) {
         try {
+            final ClientIdentifier clientId = message.getTarget();
+            final LeaderFrontendState existing = findFrontend(clientId);
+            if (existing != null) {
+                existing.touch();
+            }
+
             if (!isLeader() || !isLeaderActive()) {
                 LOG.info("{}: not currently leader, rejecting request {}. isLeader: {}, isLeaderActive: {},"
                                 + "isLeadershipTransferInProgress: {}.",
@@ -466,7 +493,15 @@ public class Shard extends RaftActor {
             }
 
             final ABIVersion selectedVersion = selectVersion(message);
-            final LeaderFrontendState frontend = getFrontend(message.getTarget());
+            final LeaderFrontendState frontend;
+            if (existing == null) {
+                frontend = new LeaderFrontendState(persistenceId(), clientId, store);
+                knownFrontends.put(clientId.getFrontendId(), frontend);
+                LOG.debug("{}: created state {} for client {}", persistenceId(), frontend, clientId);
+            } else {
+                frontend = existing;
+            }
+
             frontend.reconnect();
             message.getReplyTo().tell(new ConnectClientSuccess(message.getTarget(), message.getSequence(), getSelf(),
                 ImmutableList.of(), store.getDataTree(), CLIENT_MAX_MESSAGES).toVersion(selectedVersion),