X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=d53cb48e501c75c25e3532ce2c71e481717612ad;hb=refs%2Fchanges%2F01%2F11401%2F17;hp=770cdec39ceb430b8b1be1cbd1b4ef63ba0abbd7;hpb=c7636ed2c99ab891dbecb719db53c1a4ce1b54b8;p=controller.git
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 770cdec39c..d53cb48e50 100644
--- 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
@@ -27,9 +27,18 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.common.actor.CommonConfig;
import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
import org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.CohortEntry;
+import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort;
import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
@@ -76,14 +85,6 @@ import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;
-import javax.annotation.Nonnull;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
/**
* A Shard represents a portion of the logical data tree
*
@@ -104,10 +105,6 @@ public class Shard extends RaftActor {
private final LoggingAdapter LOG =
Logging.getLogger(getContext().system(), this);
- // By default persistent will be true and can be turned off using the system
- // property shard.persistent
- private final boolean persistent;
-
/// The name of this shard
private final ShardIdentifier name;
@@ -120,6 +117,8 @@ public class Shard extends RaftActor {
private final DatastoreContext datastoreContext;
+ private final DataPersistenceProvider dataPersistenceProvider;
+
private SchemaContext schemaContext;
private ActorRef createSnapshotTransaction;
@@ -140,20 +139,17 @@ public class Shard extends RaftActor {
private final Map transactionChains = new HashMap<>();
- protected Shard(ShardIdentifier name, Map peerAddresses,
- DatastoreContext datastoreContext, SchemaContext schemaContext) {
+ protected Shard(final ShardIdentifier name, final Map peerAddresses,
+ final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
super(name.toString(), mapPeerAddresses(peerAddresses),
Optional.of(datastoreContext.getShardRaftConfig()));
this.name = name;
this.datastoreContext = datastoreContext;
this.schemaContext = schemaContext;
+ this.dataPersistenceProvider = (datastoreContext.isPersistent()) ? new PersistentDataProvider() : new NonPersistentRaftDataProvider();
- String setting = System.getProperty("shard.persistent");
-
- this.persistent = !"false".equals(setting);
-
- LOG.info("Shard created : {} persistent : {}", name, persistent);
+ LOG.info("Shard created : {} persistent : {}", name, datastoreContext.isPersistent());
store = InMemoryDOMDataStoreFactory.create(name.toString(), null,
datastoreContext.getDataStoreProperties());
@@ -164,7 +160,6 @@ public class Shard extends RaftActor {
shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(),
datastoreContext.getDataStoreMXBeanType());
- shardMBean.setDataStoreExecutor(store.getDomStoreExecutor());
shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager());
if (isMetricsCaptureEnabled()) {
@@ -179,7 +174,7 @@ public class Shard extends RaftActor {
}
private static Map mapPeerAddresses(
- Map peerAddresses) {
+ final Map peerAddresses) {
Map map = new HashMap<>();
for (Map.Entry entry : peerAddresses
@@ -192,7 +187,7 @@ public class Shard extends RaftActor {
public static Props props(final ShardIdentifier name,
final Map peerAddresses,
- DatastoreContext datastoreContext, SchemaContext schemaContext) {
+ final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
Preconditions.checkNotNull(name, "name should not be null");
Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null");
@@ -211,7 +206,7 @@ public class Shard extends RaftActor {
}
@Override
- public void onReceiveRecover(Object message) {
+ public void onReceiveRecover(final Object message) throws Exception {
if(LOG.isDebugEnabled()) {
LOG.debug("onReceiveRecover: Received message {} from {}",
message.getClass().toString(),
@@ -230,7 +225,7 @@ public class Shard extends RaftActor {
}
@Override
- public void onReceiveCommand(Object message) {
+ public void onReceiveCommand(final Object message) throws Exception {
if(LOG.isDebugEnabled()) {
LOG.debug("onReceiveCommand: Received message {} from {}", message, getSender());
}
@@ -277,7 +272,7 @@ public class Shard extends RaftActor {
}
}
- private void handleCommitTransaction(CommitTransaction commit) {
+ private void handleCommitTransaction(final CommitTransaction commit) {
final String transactionID = commit.getTransactionID();
LOG.debug("Committing transaction {}", transactionID);
@@ -308,12 +303,8 @@ public class Shard extends RaftActor {
// currently uses a same thread executor anyway.
cohortEntry.getCohort().preCommit().get();
- if(persistent) {
- Shard.this.persistData(getSender(), transactionID,
- new CompositeModificationPayload(cohortEntry.getModification().toSerializable()));
- } else {
- Shard.this.finishCommit(getSender(), transactionID);
- }
+ Shard.this.persistData(getSender(), transactionID,
+ new CompositeModificationPayload(cohortEntry.getModification().toSerializable()));
} catch (InterruptedException | ExecutionException e) {
LOG.error(e, "An exception occurred while preCommitting transaction {}",
cohortEntry.getTransactionID());
@@ -376,13 +367,14 @@ public class Shard extends RaftActor {
commitCoordinator.currentTransactionComplete(transactionID, true);
}
- private void handleCanCommitTransaction(CanCommitTransaction canCommit) {
+ private void handleCanCommitTransaction(final CanCommitTransaction canCommit) {
LOG.debug("Can committing transaction {}", canCommit.getTransactionID());
commitCoordinator.handleCanCommit(canCommit, getSender(), self());
}
private void handleForwardedReadyTransaction(ForwardedReadyTransaction ready) {
- LOG.debug("Readying transaction {}", ready.getTransactionID());
+ LOG.debug("Readying transaction {}, client version {}", ready.getTransactionID(),
+ ready.getTxnClientVersion());
// This message is forwarded by the ShardTransaction on ready. We cache the cohort in the
// commitCoordinator in preparation for the subsequent three phase commit initiated by
@@ -390,19 +382,29 @@ public class Shard extends RaftActor {
commitCoordinator.transactionReady(ready.getTransactionID(), ready.getCohort(),
ready.getModification());
- // Return our actor path as we'll handle the three phase commit.
- ReadyTransactionReply readyTransactionReply =
- new ReadyTransactionReply(Serialization.serializedActorPath(self()));
- getSender().tell(
- ready.isReturnSerialized() ? readyTransactionReply.toSerializable() : readyTransactionReply,
- getSelf());
+ // Return our actor path as we'll handle the three phase commit, except if the Tx client
+ // version < 1 (Helium-1 version). This means the Tx was initiated by a base Helium version
+ // node. In that case, the subsequent 3-phase commit messages won't contain the
+ // transactionId so to maintain backwards compatibility, we create a separate cohort actor
+ // to provide the compatible behavior.
+ ActorRef replyActorPath = self();
+ if(ready.getTxnClientVersion() < CreateTransaction.HELIUM_1_VERSION) {
+ LOG.debug("Creating BackwardsCompatibleThreePhaseCommitCohort");
+ replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props(
+ ready.getTransactionID()));
+ }
+
+ ReadyTransactionReply readyTransactionReply = new ReadyTransactionReply(
+ Serialization.serializedActorPath(replyActorPath));
+ getSender().tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() :
+ readyTransactionReply, getSelf());
}
- private void handleAbortTransaction(AbortTransaction abort) {
+ private void handleAbortTransaction(final AbortTransaction abort) {
doAbortTransaction(abort.getTransactionID(), getSender());
}
- private void doAbortTransaction(String transactionID, final ActorRef sender) {
+ private void doAbortTransaction(final String transactionID, final ActorRef sender) {
final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
if(cohortEntry != null) {
LOG.debug("Aborting transaction {}", transactionID);
@@ -417,7 +419,7 @@ public class Shard extends RaftActor {
Futures.addCallback(future, new FutureCallback() {
@Override
- public void onSuccess(Void v) {
+ public void onSuccess(final Void v) {
shardMBean.incrementAbortTransactionsCount();
if(sender != null) {
@@ -426,7 +428,7 @@ public class Shard extends RaftActor {
}
@Override
- public void onFailure(Throwable t) {
+ public void onFailure(final Throwable t) {
LOG.error(t, "An exception happened during abort");
if(sender != null) {
@@ -437,7 +439,7 @@ public class Shard extends RaftActor {
}
}
- private void handleCreateTransaction(Object message) {
+ private void handleCreateTransaction(final Object message) {
if (isLeader()) {
createTransaction(CreateTransaction.fromSerializable(message));
} else if (getLeader() != null) {
@@ -450,7 +452,7 @@ public class Shard extends RaftActor {
}
}
- private void handleReadDataReply(Object message) {
+ private void handleReadDataReply(final Object message) {
// This must be for install snapshot. Don't want to open this up and trigger
// deSerialization
@@ -464,7 +466,7 @@ public class Shard extends RaftActor {
getSender().tell(PoisonPill.getInstance(), self());
}
- private void closeTransactionChain(CloseTransactionChain closeTransactionChain) {
+ private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
DOMStoreTransactionChain chain =
transactionChains.remove(closeTransactionChain.getTransactionChainId());
@@ -473,10 +475,8 @@ public class Shard extends RaftActor {
}
}
- private ActorRef createTypedTransactionActor(
- int transactionType,
- ShardTransactionIdentifier transactionId,
- String transactionChainId ) {
+ private ActorRef createTypedTransactionActor(int transactionType,
+ ShardTransactionIdentifier transactionId, String transactionChainId, int clientVersion ) {
DOMStoreTransactionFactory factory = store;
@@ -489,8 +489,8 @@ public class Shard extends RaftActor {
}
}
- if(this.schemaContext == null){
- throw new NullPointerException("schemaContext should not be null");
+ if(this.schemaContext == null) {
+ throw new IllegalStateException("SchemaContext is not set");
}
if (transactionType == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
@@ -500,7 +500,8 @@ public class Shard extends RaftActor {
return getContext().actorOf(
ShardTransaction.props(factory.newReadOnlyTransaction(), getSelf(),
schemaContext,datastoreContext, shardMBean,
- transactionId.getRemoteTransactionId()), transactionId.toString());
+ transactionId.getRemoteTransactionId(), clientVersion),
+ transactionId.toString());
} else if (transactionType == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
@@ -509,7 +510,8 @@ public class Shard extends RaftActor {
return getContext().actorOf(
ShardTransaction.props(factory.newReadWriteTransaction(), getSelf(),
schemaContext, datastoreContext, shardMBean,
- transactionId.getRemoteTransactionId()), transactionId.toString());
+ transactionId.getRemoteTransactionId(), clientVersion),
+ transactionId.toString());
} else if (transactionType == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
@@ -519,7 +521,8 @@ public class Shard extends RaftActor {
return getContext().actorOf(
ShardTransaction.props(factory.newWriteOnlyTransaction(), getSelf(),
schemaContext, datastoreContext, shardMBean,
- transactionId.getRemoteTransactionId()), transactionId.toString());
+ transactionId.getRemoteTransactionId(), clientVersion),
+ transactionId.toString());
} else {
throw new IllegalArgumentException(
"Shard="+name + ":CreateTransaction message has unidentified transaction type="
@@ -528,39 +531,44 @@ public class Shard extends RaftActor {
}
private void createTransaction(CreateTransaction createTransaction) {
- createTransaction(createTransaction.getTransactionType(),
- createTransaction.getTransactionId(), createTransaction.getTransactionChainId());
+ try {
+ ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(),
+ createTransaction.getTransactionId(), createTransaction.getTransactionChainId(),
+ createTransaction.getVersion());
+
+ getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor),
+ createTransaction.getTransactionId()).toSerializable(), getSelf());
+ } catch (Exception e) {
+ getSender().tell(new akka.actor.Status.Failure(e), getSelf());
+ }
}
- private ActorRef createTransaction(int transactionType, String remoteTransactionId, String transactionChainId) {
+ private ActorRef createTransaction(int transactionType, String remoteTransactionId,
+ String transactionChainId, int clientVersion) {
ShardTransactionIdentifier transactionId =
ShardTransactionIdentifier.builder()
.remoteTransactionId(remoteTransactionId)
.build();
+
if(LOG.isDebugEnabled()) {
LOG.debug("Creating transaction : {} ", transactionId);
}
- ActorRef transactionActor =
- createTypedTransactionActor(transactionType, transactionId, transactionChainId);
- getSender()
- .tell(new CreateTransactionReply(
- Serialization.serializedActorPath(transactionActor),
- remoteTransactionId).toSerializable(),
- getSelf());
+ ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId,
+ transactionChainId, clientVersion);
return transactionActor;
}
- private void syncCommitTransaction(DOMStoreWriteTransaction transaction)
+ private void syncCommitTransaction(final DOMStoreWriteTransaction transaction)
throws ExecutionException, InterruptedException {
DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
commitCohort.preCommit().get();
commitCohort.commit().get();
}
- private void commitWithNewTransaction(Modification modification) {
+ private void commitWithNewTransaction(final Modification modification) {
DOMStoreWriteTransaction tx = store.newWriteOnlyTransaction();
modification.apply(tx);
try {
@@ -573,18 +581,18 @@ public class Shard extends RaftActor {
}
}
- private void updateSchemaContext(UpdateSchemaContext message) {
+ private void updateSchemaContext(final UpdateSchemaContext message) {
this.schemaContext = message.getSchemaContext();
updateSchemaContext(message.getSchemaContext());
store.onGlobalContextUpdated(message.getSchemaContext());
}
@VisibleForTesting
- void updateSchemaContext(SchemaContext schemaContext) {
+ void updateSchemaContext(final SchemaContext schemaContext) {
store.onGlobalContextUpdated(schemaContext);
}
- private void registerChangeListener(RegisterChangeListener registerChangeListener) {
+ private void registerChangeListener(final RegisterChangeListener registerChangeListener) {
LOG.debug("registerDataChangeListener for {}", registerChangeListener.getPath());
@@ -607,12 +615,12 @@ public class Shard extends RaftActor {
LOG.debug("registerDataChangeListener sending reply, listenerRegistrationPath = {} ",
listenerRegistration.path());
- getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()),getSelf());
+ getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf());
}
private ListenerRegistration>> doChangeListenerRegistration(
- RegisterChangeListener registerChangeListener) {
+ final RegisterChangeListener registerChangeListener) {
ActorSelection dataChangeListenerPath = getContext().system().actorSelection(
registerChangeListener.getDataChangeListenerPath());
@@ -642,7 +650,7 @@ public class Shard extends RaftActor {
@Override
protected
- void startLogRecoveryBatch(int maxBatchSize) {
+ void startLogRecoveryBatch(final int maxBatchSize) {
currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize);
if(LOG.isDebugEnabled()) {
@@ -651,7 +659,7 @@ public class Shard extends RaftActor {
}
@Override
- protected void appendRecoveredLogEntry(Payload data) {
+ protected void appendRecoveredLogEntry(final Payload data) {
if (data instanceof CompositeModificationPayload) {
currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification());
} else {
@@ -660,7 +668,7 @@ public class Shard extends RaftActor {
}
@Override
- protected void applyRecoverySnapshot(ByteString snapshot) {
+ protected void applyRecoverySnapshot(final ByteString snapshot) {
if(recoveryCoordinator == null) {
recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext);
}
@@ -725,7 +733,7 @@ public class Shard extends RaftActor {
}
@Override
- protected void applyState(ActorRef clientActor, String identifier, Object data) {
+ protected void applyState(final ActorRef clientActor, final String identifier, final Object data) {
if (data instanceof CompositeModificationPayload) {
Object modification = ((CompositeModificationPayload) data).getModification();
@@ -773,7 +781,8 @@ public class Shard extends RaftActor {
// so that this actor does not get block building the snapshot
createSnapshotTransaction = createTransaction(
TransactionProxy.TransactionType.READ_ONLY.ordinal(),
- "createSnapshot" + ++createSnapshotTransactionCounter, "");
+ "createSnapshot" + ++createSnapshotTransactionCounter, "",
+ CreateTransaction.CURRENT_VERSION);
createSnapshotTransaction.tell(
new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());
@@ -783,7 +792,7 @@ public class Shard extends RaftActor {
@VisibleForTesting
@Override
- protected void applySnapshot(ByteString snapshot) {
+ protected void applySnapshot(final ByteString snapshot) {
// Since this will be done only on Recovery or when this actor is a Follower
// we can safely commit everything in here. We not need to worry about event notifications
// as they would have already been disabled on the follower
@@ -843,7 +852,12 @@ public class Shard extends RaftActor {
}
}
- @Override protected void onLeaderChanged(String oldLeader, String newLeader) {
+ @Override
+ protected DataPersistenceProvider persistence() {
+ return dataPersistenceProvider;
+ }
+
+ @Override protected void onLeaderChanged(final String oldLeader, final String newLeader) {
shardMBean.setLeader(newLeader);
}
@@ -851,6 +865,11 @@ public class Shard extends RaftActor {
return this.name.toString();
}
+ @VisibleForTesting
+ DataPersistenceProvider getDataPersistenceProvider() {
+ return dataPersistenceProvider;
+ }
+
private static class ShardCreator implements Creator {
private static final long serialVersionUID = 1L;
@@ -860,8 +879,8 @@ public class Shard extends RaftActor {
final DatastoreContext datastoreContext;
final SchemaContext schemaContext;
- ShardCreator(ShardIdentifier name, Map peerAddresses,
- DatastoreContext datastoreContext, SchemaContext schemaContext) {
+ ShardCreator(final ShardIdentifier name, final Map peerAddresses,
+ final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
this.name = name;
this.peerAddresses = peerAddresses;
this.datastoreContext = datastoreContext;
@@ -894,11 +913,11 @@ public class Shard extends RaftActor {
private volatile ListenerRegistration>> delegate;
- DelayedListenerRegistration(RegisterChangeListener registerChangeListener) {
+ DelayedListenerRegistration(final RegisterChangeListener registerChangeListener) {
this.registerChangeListener = registerChangeListener;
}
- void setDelegate( ListenerRegistration>> registration) {
this.delegate = registration;
}