Expose stream version used for DataTreeCandidate stream

We need to properly version the payload stream, as it gives us
the view into which version produced it. This allows us to make
proper typing assumptions in normalization (and thus pruning).

JIRA: CONTROLLER-1923
Change-Id: Iaea995a5ac58e1fba2a4199d3355b321cf9fcff3
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java
index 016b078d5c002f1154aa6908920f4f27c65a2c52..bbcc42c3b83def4aebf2dde02301006081b95c16 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java
@@ -57,6 +57,7 @@ import org.opendaylight.controller.cluster.datastore.persisted.AbstractIdentifia
 import org.opendaylight.controller.cluster.datastore.persisted.CloseLocalHistoryPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.CreateLocalHistoryPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.CloseLocalHistoryPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.CreateLocalHistoryPayload;

+import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;

 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
 import org.opendaylight.controller.cluster.datastore.persisted.PurgeLocalHistoryPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.PurgeTransactionPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
 import org.opendaylight.controller.cluster.datastore.persisted.PurgeLocalHistoryPayload;
 import org.opendaylight.controller.cluster.datastore.persisted.PurgeTransactionPayload;


@@ -316,10 +317,11 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
 
     @SuppressWarnings("checkstyle:IllegalCatch")
     private void applyRecoveryCandidate(final CommitTransactionPayload payload) throws IOException {
 
     @SuppressWarnings("checkstyle:IllegalCatch")
     private void applyRecoveryCandidate(final CommitTransactionPayload payload) throws IOException {

-        final Entry<TransactionIdentifier, DataTreeCandidate> entry = payload.getCandidate();
+        final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> entry = payload.getCandidate();

         final DataTreeModification unwrapped = dataTree.takeSnapshot().newModification();
         final DataTreeModification unwrapped = dataTree.takeSnapshot().newModification();

+        // FIXME: CONTROLLER-1923: examine version first

         final PruningDataTreeModification mod = wrapWithPruning(unwrapped);
         final PruningDataTreeModification mod = wrapWithPruning(unwrapped);

-        DataTreeCandidates.applyToModification(mod, entry.getValue());
+        DataTreeCandidates.applyToModification(mod, entry.getValue().getCandidate());

         mod.ready();
 
         LOG.trace("{}: Applying recovery modification {}", logContext, unwrapped);
         mod.ready();
 
         LOG.trace("{}: Applying recovery modification {}", logContext, unwrapped);


@@ -367,12 +369,13 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
 
     private void applyReplicatedCandidate(final CommitTransactionPayload payload)
             throws DataValidationFailedException, IOException {
 
     private void applyReplicatedCandidate(final CommitTransactionPayload payload)
             throws DataValidationFailedException, IOException {

-        final Entry<TransactionIdentifier, DataTreeCandidate> entry = payload.getCandidate();
+        final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> entry = payload.getCandidate();

         final TransactionIdentifier identifier = entry.getKey();
         LOG.debug("{}: Applying foreign transaction {}", logContext, identifier);
 
         final DataTreeModification mod = dataTree.takeSnapshot().newModification();
         final TransactionIdentifier identifier = entry.getKey();
         LOG.debug("{}: Applying foreign transaction {}", logContext, identifier);
 
         final DataTreeModification mod = dataTree.takeSnapshot().newModification();

-        DataTreeCandidates.applyToModification(mod, entry.getValue());
+        // TODO: check version here, which will enable us to perform forward-compatibility transformations
+        DataTreeCandidates.applyToModification(mod, entry.getValue().getCandidate());

         mod.ready();
 
         LOG.trace("{}: Applying foreign modification {}", logContext, mod);
         mod.ready();
 
         LOG.trace("{}: Applying foreign modification {}", logContext, mod);

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java
index b2164133b5b1209c53484852414bf88bc84a43a1..fed348320defd0494318956c5f2e9d847e773af9 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java
@@ -27,6 +27,7 @@ import java.io.StreamCorruptedException;
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.Map.Entry;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.Map.Entry;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;

+import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;

 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.IdentifiablePayload;
 import org.opendaylight.yangtools.concepts.Variant;
 import org.opendaylight.yangtools.yang.data.api.schema.stream.ReusableStreamReceiver;
 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.IdentifiablePayload;
 import org.opendaylight.yangtools.concepts.Variant;
 import org.opendaylight.yangtools.yang.data.api.schema.stream.ReusableStreamReceiver;


@@ -47,7 +48,7 @@ public abstract class CommitTransactionPayload extends IdentifiablePayload<Trans
     private static final Logger LOG = LoggerFactory.getLogger(CommitTransactionPayload.class);
     private static final long serialVersionUID = 1L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CommitTransactionPayload.class);
     private static final long serialVersionUID = 1L;
 

-    private volatile Entry<TransactionIdentifier, DataTreeCandidate> candidate = null;
+    private volatile Entry<TransactionIdentifier, DataTreeCandidateWithVersion> candidate = null;

 
     CommitTransactionPayload() {
 
 
     CommitTransactionPayload() {
 


@@ -73,8 +74,8 @@ public abstract class CommitTransactionPayload extends IdentifiablePayload<Trans
         return create(transactionId, candidate, 512);
     }
 
         return create(transactionId, candidate, 512);
     }
 

-    public Entry<TransactionIdentifier, DataTreeCandidate> getCandidate() throws IOException {
-        Entry<TransactionIdentifier, DataTreeCandidate> localCandidate = candidate;
+    public Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate() throws IOException {
+        Entry<TransactionIdentifier, DataTreeCandidateWithVersion> localCandidate = candidate;

         if (localCandidate == null) {
             synchronized (this) {
                 localCandidate = candidate;
         if (localCandidate == null) {
             synchronized (this) {
                 localCandidate = candidate;


@@ -86,13 +87,14 @@ public abstract class CommitTransactionPayload extends IdentifiablePayload<Trans
         return localCandidate;
     }
 
         return localCandidate;
     }
 

-    public final Entry<TransactionIdentifier, DataTreeCandidate> getCandidate(
+    public final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate(

             final ReusableStreamReceiver receiver) throws IOException {
         final DataInput in = newDataInput();
         return new SimpleImmutableEntry<>(TransactionIdentifier.readFrom(in),
                 DataTreeCandidateInputOutput.readDataTreeCandidate(in, receiver));
     }
 
             final ReusableStreamReceiver receiver) throws IOException {
         final DataInput in = newDataInput();
         return new SimpleImmutableEntry<>(TransactionIdentifier.readFrom(in),
                 DataTreeCandidateInputOutput.readDataTreeCandidate(in, receiver));
     }
 

+    @Override

     public TransactionIdentifier getIdentifier() {
         try  {
             return getCandidate().getKey();
     public TransactionIdentifier getIdentifier() {
         try  {
             return getCandidate().getKey();

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/DataTreeCandidateInputOutput.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/DataTreeCandidateInputOutput.java
index 60d70e44c8af09b8df50c8d3c4d34025f6b65f23..b78bb563aff7634e3c963599b745f36ee96ef3ef 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/DataTreeCandidateInputOutput.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/DataTreeCandidateInputOutput.java
@@ -7,6 +7,8 @@
  */
 package org.opendaylight.controller.cluster.datastore.persisted;
 
  */
 package org.opendaylight.controller.cluster.datastore.persisted;
 

+import static java.util.Objects.requireNonNull;
+

 import com.google.common.annotations.Beta;
 import com.google.common.collect.ImmutableList;
 import java.io.DataInput;
 import com.google.common.annotations.Beta;
 import com.google.common.collect.ImmutableList;
 import java.io.DataInput;


@@ -14,6 +16,8 @@ import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;

+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.opendaylight.yangtools.concepts.Immutable;

 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
 import org.opendaylight.yangtools.yang.data.api.schema.stream.ReusableStreamReceiver;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
 import org.opendaylight.yangtools.yang.data.api.schema.stream.ReusableStreamReceiver;


@@ -24,6 +28,7 @@ import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataInput;
 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataInput;
 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput;

+import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 


@@ -97,8 +102,28 @@ public final class DataTreeCandidateInputOutput {
         }
     }
 
         }
     }
 

-    public static DataTreeCandidate readDataTreeCandidate(final DataInput in, final ReusableStreamReceiver receiver)
-            throws IOException {
+    @NonNullByDefault
+    public static final class DataTreeCandidateWithVersion implements Immutable {
+        private final DataTreeCandidate candidate;
+        private final NormalizedNodeStreamVersion version;
+
+        public DataTreeCandidateWithVersion(final DataTreeCandidate candidate,
+                final NormalizedNodeStreamVersion version) {
+            this.candidate = requireNonNull(candidate);
+            this.version = requireNonNull(version);
+        }
+
+        public DataTreeCandidate getCandidate() {
+            return candidate;
+        }
+
+        public NormalizedNodeStreamVersion getVersion() {
+            return version;
+        }
+    }
+
+    public static DataTreeCandidateWithVersion readDataTreeCandidate(final DataInput in,
+            final ReusableStreamReceiver receiver) throws IOException {

         final NormalizedNodeDataInput reader = NormalizedNodeDataInput.newDataInput(in);
         final YangInstanceIdentifier rootPath = reader.readYangInstanceIdentifier();
         final byte type = reader.readByte();
         final NormalizedNodeDataInput reader = NormalizedNodeDataInput.newDataInput(in);
         final YangInstanceIdentifier rootPath = reader.readYangInstanceIdentifier();
         final byte type = reader.readByte();


@@ -130,7 +155,8 @@ public final class DataTreeCandidateInputOutput {
                 throw new IllegalArgumentException("Unhandled node type " + type);
         }
 
                 throw new IllegalArgumentException("Unhandled node type " + type);
         }
 

-        return DataTreeCandidates.newDataTreeCandidate(rootPath, rootNode);
+        return new DataTreeCandidateWithVersion(DataTreeCandidates.newDataTreeCandidate(rootPath, rootNode),
+            reader.getVersion());

     }
 
     private static void writeChildren(final NormalizedNodeDataOutput out,
     }
 
     private static void writeChildren(final NormalizedNodeDataOutput out,

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayloadTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayloadTest.java
index e0518a791bc0051f379cafc0b908c62c95518e05..c081b89137e154b2d5d5ab3b7338b1da23b62a71 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayloadTest.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayloadTest.java
@@ -18,6 +18,7 @@ import org.apache.commons.lang3.SerializationUtils;
 import org.junit.Before;
 import org.junit.Test;
 import org.opendaylight.controller.cluster.datastore.AbstractTest;
 import org.junit.Before;
 import org.junit.Test;
 import org.opendaylight.controller.cluster.datastore.AbstractTest;

+import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;

 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
 import org.opendaylight.yangtools.yang.common.QName;
 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
 import org.opendaylight.yangtools.yang.common.QName;


@@ -67,9 +68,11 @@ public class CommitTransactionPayloadTest extends AbstractTest {
         }
     }
 
         }
     }
 

-    private static void assertCandidateEquals(final DataTreeCandidate expected, final DataTreeCandidate actual) {
-        assertEquals("root path", expected.getRootPath(), actual.getRootPath());
-        assertCandidateNodeEquals(expected.getRootNode(), actual.getRootNode());
+    private static void assertCandidateEquals(final DataTreeCandidate expected,
+            final DataTreeCandidateWithVersion actual) {
+        final DataTreeCandidate candidate = actual.getCandidate();
+        assertEquals("root path", expected.getRootPath(), candidate.getRootPath());
+        assertCandidateNodeEquals(expected.getRootNode(), candidate.getRootNode());

     }
 
     private static void assertCandidateNodeEquals(final DataTreeCandidateNode expected,
     }
 
     private static void assertCandidateNodeEquals(final DataTreeCandidateNode expected,