X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=a22e535fad1f6fa7052121c6be736a299e0b48cb;hp=5ea9b30c63e7d3f9b44e9800eddb2b5e42f08657;hb=20b2ba67a79b1ec7f3d2fd58ed32f22ea54a1b59;hpb=f281dc3e886501dc617e397c8e203732300a8f08
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 5ea9b30c63..a22e535fad 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,6 +27,13 @@ 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;
@@ -58,6 +65,7 @@ import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContex
import org.opendaylight.controller.cluster.datastore.modification.Modification;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
+import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier;
import org.opendaylight.controller.cluster.raft.RaftActor;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
@@ -78,14 +86,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
*
@@ -132,6 +132,8 @@ public class Shard extends RaftActor {
private Cancellable txCommitTimeoutCheckSchedule;
+ private Optional roleChangeNotifier;
+
/**
* Coordinates persistence recovery on startup.
*/
@@ -140,8 +142,8 @@ 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()));
@@ -161,7 +163,6 @@ public class Shard extends RaftActor {
shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(),
datastoreContext.getDataStoreMXBeanType());
- shardMBean.setDataStoreExecutor(store.getDomStoreExecutor());
shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager());
if (isMetricsCaptureEnabled()) {
@@ -173,10 +174,13 @@ public class Shard extends RaftActor {
transactionCommitTimeout = TimeUnit.MILLISECONDS.convert(
datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS);
+
+ // create a notifier actor for each cluster member
+ roleChangeNotifier = createRoleChangeNotifier(name.toString());
}
private static Map mapPeerAddresses(
- Map peerAddresses) {
+ final Map peerAddresses) {
Map map = new HashMap<>();
for (Map.Entry entry : peerAddresses
@@ -189,7 +193,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");
@@ -198,6 +202,12 @@ public class Shard extends RaftActor {
return Props.create(new ShardCreator(name, peerAddresses, datastoreContext, schemaContext));
}
+ private Optional createRoleChangeNotifier(String shardId) {
+ ActorRef shardRoleChangeNotifier = this.getContext().actorOf(
+ RoleChangeNotifier.getProps(shardId), shardId + "-notifier");
+ return Optional.of(shardRoleChangeNotifier);
+ }
+
@Override
public void postStop() {
super.postStop();
@@ -208,7 +218,7 @@ public class Shard extends RaftActor {
}
@Override
- public void onReceiveRecover(Object message) throws Exception {
+ public void onReceiveRecover(final Object message) throws Exception {
if(LOG.isDebugEnabled()) {
LOG.debug("onReceiveRecover: Received message {} from {}",
message.getClass().toString(),
@@ -227,7 +237,7 @@ public class Shard extends RaftActor {
}
@Override
- public void onReceiveCommand(Object message) throws Exception {
+ public void onReceiveCommand(final Object message) throws Exception {
if(LOG.isDebugEnabled()) {
LOG.debug("onReceiveCommand: Received message {} from {}", message, getSender());
}
@@ -261,6 +271,11 @@ public class Shard extends RaftActor {
}
}
+ @Override
+ protected Optional getRoleChangeNotifier() {
+ return roleChangeNotifier;
+ }
+
private void handleTransactionCommitTimeoutCheck() {
CohortEntry cohortEntry = commitCoordinator.getCurrentCohortEntry();
if(cohortEntry != null) {
@@ -274,7 +289,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);
@@ -369,7 +384,7 @@ 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());
}
@@ -402,11 +417,11 @@ public class Shard extends RaftActor {
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);
@@ -421,7 +436,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) {
@@ -430,7 +445,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) {
@@ -441,7 +456,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) {
@@ -454,7 +469,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
@@ -468,7 +483,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());
@@ -491,8 +506,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()) {
@@ -533,9 +548,16 @@ public class Shard extends RaftActor {
}
private void createTransaction(CreateTransaction createTransaction) {
- createTransaction(createTransaction.getTransactionType(),
- createTransaction.getTransactionId(), createTransaction.getTransactionChainId(),
- createTransaction.getVersion());
+ 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,
@@ -545,29 +567,25 @@ public class Shard extends RaftActor {
ShardTransactionIdentifier.builder()
.remoteTransactionId(remoteTransactionId)
.build();
+
if(LOG.isDebugEnabled()) {
LOG.debug("Creating transaction : {} ", transactionId);
}
+
ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId,
transactionChainId, clientVersion);
- getSender()
- .tell(new CreateTransactionReply(
- Serialization.serializedActorPath(transactionActor),
- remoteTransactionId).toSerializable(),
- getSelf());
-
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 {
@@ -580,18 +598,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());
@@ -619,7 +637,7 @@ public class Shard extends RaftActor {
private ListenerRegistration>> doChangeListenerRegistration(
- RegisterChangeListener registerChangeListener) {
+ final RegisterChangeListener registerChangeListener) {
ActorSelection dataChangeListenerPath = getContext().system().actorSelection(
registerChangeListener.getDataChangeListenerPath());
@@ -649,7 +667,7 @@ public class Shard extends RaftActor {
@Override
protected
- void startLogRecoveryBatch(int maxBatchSize) {
+ void startLogRecoveryBatch(final int maxBatchSize) {
currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize);
if(LOG.isDebugEnabled()) {
@@ -658,7 +676,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 {
@@ -667,7 +685,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);
}
@@ -732,7 +750,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();
@@ -770,6 +788,7 @@ public class Shard extends RaftActor {
shardMBean.setCommitIndex(getCommitIndex());
shardMBean.setLastApplied(getLastApplied());
+ shardMBean.setInMemoryJournalDataSize(getRaftActorContext().getReplicatedLog().dataSize());
}
@Override
@@ -791,7 +810,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
@@ -856,7 +875,7 @@ public class Shard extends RaftActor {
return dataPersistenceProvider;
}
- @Override protected void onLeaderChanged(String oldLeader, String newLeader) {
+ @Override protected void onLeaderChanged(final String oldLeader, final String newLeader) {
shardMBean.setLeader(newLeader);
}
@@ -878,8 +897,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;
@@ -912,11 +931,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;
}