+ private boolean isMetricsCaptureEnabled(){
+ CommonConfig config = new CommonConfig(getContext().system().settings().config());
+ return config.isMetricCaptureEnabled();
+ }
+
+ @Override
+ protected
+ void startLogRecoveryBatch(int maxBatchSize) {
+ currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize);
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("{} : starting log recovery batch with max size {}", persistenceId(), maxBatchSize);
+ }
+ }
+
+ @Override
+ protected void appendRecoveredLogEntry(Payload data) {
+ if (data instanceof CompositeModificationPayload) {
+ currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification());
+ } else {
+ LOG.error("Unknown state received {} during recovery", data);
+ }
+ }
+
+ @Override
+ protected void applyRecoverySnapshot(ByteString snapshot) {
+ if(recoveryCoordinator == null) {
+ recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext);
+ }
+
+ recoveryCoordinator.submit(snapshot, store.newWriteOnlyTransaction());
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("{} : submitted recovery sbapshot", persistenceId());
+ }
+ }
+
+ @Override
+ protected void applyCurrentLogRecoveryBatch() {
+ if(recoveryCoordinator == null) {
+ recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext);
+ }
+
+ recoveryCoordinator.submit(currentLogRecoveryBatch, store.newWriteOnlyTransaction());
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("{} : submitted log recovery batch with size {}", persistenceId(),
+ currentLogRecoveryBatch.size());
+ }
+ }
+
+ @Override
+ protected void onRecoveryComplete() {
+ if(recoveryCoordinator != null) {
+ Collection<DOMStoreWriteTransaction> txList = recoveryCoordinator.getTransactions();
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("{} : recovery complete - committing {} Tx's", persistenceId(), txList.size());
+ }
+
+ for(DOMStoreWriteTransaction tx: txList) {
+ try {
+ syncCommitTransaction(tx);
+ shardMBean.incrementCommittedTransactionCount();
+ } catch (InterruptedException | ExecutionException e) {
+ shardMBean.incrementFailedTransactionsCount();
+ LOG.error(e, "Failed to commit");
+ }
+ }
+ }
+
+ recoveryCoordinator = null;
+ currentLogRecoveryBatch = null;
+ updateJournalStats();
+
+ //notify shard manager
+ getContext().parent().tell(new ActorInitialized(), getSelf());
+
+ // Schedule a message to be periodically sent to check if the current in-progress
+ // transaction should be expired and aborted.
+ FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
+ txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
+ period, period, getSelf(),
+ TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
+ }
+
+ @Override
+ protected void applyState(ActorRef clientActor, String identifier, Object data) {
+
+ if (data instanceof CompositeModificationPayload) {
+ Object modification = ((CompositeModificationPayload) data).getModification();
+
+ if(modification == null) {
+ LOG.error(
+ "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
+ identifier, clientActor != null ? clientActor.path().toString() : null);
+ } else if(clientActor == null) {
+ // There's no clientActor to which to send a commit reply so we must be applying
+ // replicated state from the leader.
+ commitWithNewTransaction(MutableCompositeModification.fromSerializable(
+ modification, schemaContext));
+ } else {
+ // This must be the OK to commit after replication consensus.
+ finishCommit(clientActor, identifier);
+ }
+ } else {
+ LOG.error("Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}",
+ data, data.getClass().getClassLoader(),
+ CompositeModificationPayload.class.getClassLoader());
+ }
+
+ updateJournalStats();
+
+ }
+
+ private void updateJournalStats() {
+ ReplicatedLogEntry lastLogEntry = getLastLogEntry();
+
+ if (lastLogEntry != null) {
+ shardMBean.setLastLogIndex(lastLogEntry.getIndex());
+ shardMBean.setLastLogTerm(lastLogEntry.getTerm());
+ }
+
+ shardMBean.setCommitIndex(getCommitIndex());
+ shardMBean.setLastApplied(getLastApplied());
+ }
+
+ @Override
+ protected void createSnapshot() {
+ if (createSnapshotTransaction == null) {
+
+ // Create a transaction. We are really going to treat the transaction as a worker
+ // so that this actor does not get block building the snapshot
+ createSnapshotTransaction = createTransaction(
+ TransactionProxy.TransactionType.READ_ONLY.ordinal(),
+ "createSnapshot" + ++createSnapshotTransactionCounter, "");
+
+ createSnapshotTransaction.tell(
+ new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());
+
+ }
+ }
+
+ @VisibleForTesting
+ @Override
+ protected void applySnapshot(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
+
+ LOG.info("Applying snapshot");
+ try {
+ DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
+ NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot);
+ NormalizedNode<?, ?> node = new NormalizedNodeToNodeCodec(schemaContext)
+ .decode(serializedNode);
+
+ // delete everything first
+ transaction.delete(YangInstanceIdentifier.builder().build());
+
+ // Add everything from the remote node back
+ transaction.write(YangInstanceIdentifier.builder().build(), node);
+ syncCommitTransaction(transaction);
+ } catch (InvalidProtocolBufferException | InterruptedException | ExecutionException e) {
+ LOG.error(e, "An exception occurred when applying snapshot");
+ } finally {
+ LOG.info("Done applying snapshot");
+ }
+ }
+
+ @Override protected void onStateChanged() {
+ for (ActorSelection dataChangeListener : dataChangeListeners) {
+ dataChangeListener
+ .tell(new EnableNotification(isLeader()), getSelf());
+ }
+
+ shardMBean.setRaftState(getRaftState().name());
+ shardMBean.setCurrentTerm(getCurrentTerm());
+
+ // If this actor is no longer the leader close all the transaction chains
+ if(!isLeader()){
+ for(Map.Entry<String, DOMStoreTransactionChain> entry : transactionChains.entrySet()){
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(
+ "onStateChanged: Closing transaction chain {} because shard {} is no longer the leader",
+ entry.getKey(), getId());
+ }
+ entry.getValue().close();
+ }
+
+ transactionChains.clear();
+ }
+ }
+
+ @Override protected void onLeaderChanged(String oldLeader, String newLeader) {
+ shardMBean.setLeader(newLeader);
+ }
+
+ @Override public String persistenceId() {
+ return this.name.toString();
+ }
+
+ private static class ShardCreator implements Creator<Shard> {
+
+ private static final long serialVersionUID = 1L;
+
+ final ShardIdentifier name;
+ final Map<ShardIdentifier, String> peerAddresses;
+ final DatastoreContext datastoreContext;
+ final SchemaContext schemaContext;
+
+ ShardCreator(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
+ DatastoreContext datastoreContext, SchemaContext schemaContext) {
+ this.name = name;
+ this.peerAddresses = peerAddresses;
+ this.datastoreContext = datastoreContext;
+ this.schemaContext = schemaContext;
+ }
+
+ @Override
+ public Shard create() throws Exception {
+ return new Shard(name, peerAddresses, datastoreContext, schemaContext);
+ }
+ }
+
+ @VisibleForTesting
+ InMemoryDOMDataStore getDataStore() {
+ return store;
+ }
+
+ @VisibleForTesting
+ ShardStats getShardMBean() {
+ return shardMBean;