BUG-8618: make sync threshold tuneable

We are observing quite a few of these transitions, which may be coming
from batching scenarios. Introduce sync-index-threshold config knob
to expose control over it.

Change-Id: Ief4c89c2fe5b95cebaf3fb83cbcdda37cac126b6
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>
(cherry picked from commit 890e4bbf40aee318a2174bd4130cf34437e5617b)

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java
index 86ce3113fab53e9f66d64db0fa4d793fd08fc010..8d6b965d9bc7950727b47104da5b9a7db42dc3f1 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java
@@ -131,4 +131,12 @@ public interface ConfigParams {
      * @return the threshold in terms of number of bytes.
      */
     int getFileBackedStreamingThreshold();
      * @return the threshold in terms of number of bytes.
      */
     int getFileBackedStreamingThreshold();

+
+    /**
+     * Returns the threshold in terms of number journal entries that we can lag behind a leader until we raise a
+     * 'not synced' transition.
+     *
+     * @return the threshold in terms of number of journal entries.
+     */
+    long getSyncIndexThreshold();

 }
 }

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java
index 56fb63367207b94ad368214a569e2009f8c5648f..64502858a754b294c8f3b2d2aca31e056639f9f4 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java
@@ -49,6 +49,8 @@ public class DefaultConfigParamsImpl implements ConfigParams {
     public static final FiniteDuration HEART_BEAT_INTERVAL =
         new FiniteDuration(100, TimeUnit.MILLISECONDS);
 
     public static final FiniteDuration HEART_BEAT_INTERVAL =
         new FiniteDuration(100, TimeUnit.MILLISECONDS);
 

+    private final Supplier<RaftPolicy> policySupplier = Suppliers.memoize(this::getPolicy);
+

     private FiniteDuration heartBeatInterval = HEART_BEAT_INTERVAL;
     private long snapshotBatchCount = SNAPSHOT_BATCH_COUNT;
     private int journalRecoveryLogBatchSize = JOURNAL_RECOVERY_LOG_BATCH_SIZE;
     private FiniteDuration heartBeatInterval = HEART_BEAT_INTERVAL;
     private long snapshotBatchCount = SNAPSHOT_BATCH_COUNT;
     private int journalRecoveryLogBatchSize = JOURNAL_RECOVERY_LOG_BATCH_SIZE;


@@ -64,53 +66,53 @@ public class DefaultConfigParamsImpl implements ConfigParams {
     private long electionTimeoutFactor = 2;
     private String customRaftPolicyImplementationClass;
 
     private long electionTimeoutFactor = 2;
     private String customRaftPolicyImplementationClass;
 

-    private final Supplier<RaftPolicy> policySupplier = Suppliers.memoize(new PolicySupplier());
-

     private PeerAddressResolver peerAddressResolver = NoopPeerAddressResolver.INSTANCE;
 
     private String tempFileDirectory = "";
 
     private int fileBackedStreamingThreshold = 128 * MEGABYTE;
 
     private PeerAddressResolver peerAddressResolver = NoopPeerAddressResolver.INSTANCE;
 
     private String tempFileDirectory = "";
 
     private int fileBackedStreamingThreshold = 128 * MEGABYTE;
 

-    public void setHeartBeatInterval(FiniteDuration heartBeatInterval) {
+    private long syncIndexThreshold = 10;
+
+    public void setHeartBeatInterval(final FiniteDuration heartBeatInterval) {

         this.heartBeatInterval = heartBeatInterval;
         electionTimeOutInterval = null;
     }
 
         this.heartBeatInterval = heartBeatInterval;
         electionTimeOutInterval = null;
     }
 

-    public void setSnapshotBatchCount(long snapshotBatchCount) {
+    public void setSnapshotBatchCount(final long snapshotBatchCount) {

         this.snapshotBatchCount = snapshotBatchCount;
     }
 
         this.snapshotBatchCount = snapshotBatchCount;
     }
 

-    public void setSnapshotDataThresholdPercentage(int snapshotDataThresholdPercentage) {
+    public void setSnapshotDataThresholdPercentage(final int snapshotDataThresholdPercentage) {

         this.snapshotDataThresholdPercentage = snapshotDataThresholdPercentage;
     }
 
         this.snapshotDataThresholdPercentage = snapshotDataThresholdPercentage;
     }
 

-    public void setSnapshotChunkSize(int snapshotChunkSize) {
+    public void setSnapshotChunkSize(final int snapshotChunkSize) {

         this.snapshotChunkSize = snapshotChunkSize;
     }
 
         this.snapshotChunkSize = snapshotChunkSize;
     }
 

-    public void setJournalRecoveryLogBatchSize(int journalRecoveryLogBatchSize) {
+    public void setJournalRecoveryLogBatchSize(final int journalRecoveryLogBatchSize) {

         this.journalRecoveryLogBatchSize = journalRecoveryLogBatchSize;
     }
 
         this.journalRecoveryLogBatchSize = journalRecoveryLogBatchSize;
     }
 

-    public void setIsolatedLeaderCheckInterval(FiniteDuration isolatedLeaderCheckInterval) {
+    public void setIsolatedLeaderCheckInterval(final FiniteDuration isolatedLeaderCheckInterval) {

         this.isolatedLeaderCheckInterval = isolatedLeaderCheckInterval.toMillis();
     }
 
         this.isolatedLeaderCheckInterval = isolatedLeaderCheckInterval.toMillis();
     }
 

-    public void setElectionTimeoutFactor(long electionTimeoutFactor) {
+    public void setElectionTimeoutFactor(final long electionTimeoutFactor) {

         this.electionTimeoutFactor = electionTimeoutFactor;
         electionTimeOutInterval = null;
     }
 
         this.electionTimeoutFactor = electionTimeoutFactor;
         electionTimeOutInterval = null;
     }
 

-    public void setTempFileDirectory(String tempFileDirectory) {
+    public void setTempFileDirectory(final String tempFileDirectory) {

         this.tempFileDirectory = tempFileDirectory;
     }
 
         this.tempFileDirectory = tempFileDirectory;
     }
 

-    public void setFileBackedStreamingThreshold(int fileBackedStreamingThreshold) {
+    public void setFileBackedStreamingThreshold(final int fileBackedStreamingThreshold) {

         this.fileBackedStreamingThreshold = fileBackedStreamingThreshold;
     }
 
         this.fileBackedStreamingThreshold = fileBackedStreamingThreshold;
     }
 

-    public void setCustomRaftPolicyImplementationClass(String customRaftPolicyImplementationClass) {
+    public void setCustomRaftPolicyImplementationClass(final String customRaftPolicyImplementationClass) {

         this.customRaftPolicyImplementationClass = customRaftPolicyImplementationClass;
     }
 
         this.customRaftPolicyImplementationClass = customRaftPolicyImplementationClass;
     }
 


@@ -184,37 +186,45 @@ public class DefaultConfigParamsImpl implements ConfigParams {
         return fileBackedStreamingThreshold;
     }
 
         return fileBackedStreamingThreshold;
     }
 

-    private class PolicySupplier implements Supplier<RaftPolicy> {
-        @Override
-        @SuppressWarnings("checkstyle:IllegalCatch")
-        public RaftPolicy get() {
-            if (Strings.isNullOrEmpty(DefaultConfigParamsImpl.this.customRaftPolicyImplementationClass)) {
-                LOG.debug("No custom RaftPolicy specified. Using DefaultRaftPolicy");
-                return DefaultRaftPolicy.INSTANCE;
-            }
-
-            try {
-                String className = DefaultConfigParamsImpl.this.customRaftPolicyImplementationClass;
-                LOG.info("Trying to use custom RaftPolicy {}", className);
-                return (RaftPolicy)Class.forName(className).newInstance();
-            } catch (Exception e) {
-                if (LOG.isDebugEnabled()) {
-                    LOG.error("Could not create custom raft policy, will stick with default", e);
-                } else {
-                    LOG.error("Could not create custom raft policy, will stick with default : cause = {}",
-                            e.getMessage());
-                }
-            }
-            return DefaultRaftPolicy.INSTANCE;
-        }
-    }

 
     @Override
     public PeerAddressResolver getPeerAddressResolver() {
         return peerAddressResolver;
     }
 
 
     @Override
     public PeerAddressResolver getPeerAddressResolver() {
         return peerAddressResolver;
     }
 

-    public void setPeerAddressResolver(@Nonnull PeerAddressResolver peerAddressResolver) {
+    public void setPeerAddressResolver(@Nonnull final PeerAddressResolver peerAddressResolver) {

         this.peerAddressResolver = Preconditions.checkNotNull(peerAddressResolver);
     }
         this.peerAddressResolver = Preconditions.checkNotNull(peerAddressResolver);
     }

+
+    @Override
+    public long getSyncIndexThreshold() {
+        return syncIndexThreshold;
+    }
+
+    public void setSyncIndexThreshold(final long syncIndexThreshold) {
+        Preconditions.checkArgument(syncIndexThreshold >= 0);
+        this.syncIndexThreshold = syncIndexThreshold;
+    }
+
+    @SuppressWarnings("checkstyle:IllegalCatch")
+    private RaftPolicy getPolicy() {
+        if (Strings.isNullOrEmpty(DefaultConfigParamsImpl.this.customRaftPolicyImplementationClass)) {
+            LOG.debug("No custom RaftPolicy specified. Using DefaultRaftPolicy");
+            return DefaultRaftPolicy.INSTANCE;
+        }
+
+        try {
+            String className = DefaultConfigParamsImpl.this.customRaftPolicyImplementationClass;
+            LOG.info("Trying to use custom RaftPolicy {}", className);
+            return (RaftPolicy)Class.forName(className).newInstance();
+        } catch (Exception e) {
+            if (LOG.isDebugEnabled()) {
+                LOG.error("Could not create custom raft policy, will stick with default", e);
+            } else {
+                LOG.error("Could not create custom raft policy, will stick with default : cause = {}",
+                    e.getMessage());
+            }
+        }
+        return DefaultRaftPolicy.INSTANCE;
+    }

 }
 }

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
index b512089692eed80da136bc8f7dfc2f5a2e10274c..c35de820dba734d8c48ad9493c372c4cf0d510cb 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
@@ -51,8 +51,6 @@ import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  * </ul>
  */
 public class Follower extends AbstractRaftActorBehavior {
  * </ul>
  */
 public class Follower extends AbstractRaftActorBehavior {

-    private static final int SYNC_THRESHOLD = 10;
-

     private static final long MAX_ELECTION_TIMEOUT_FACTOR = 18;
 
     private final SyncStatusTracker initialSyncStatusTracker;
     private static final long MAX_ELECTION_TIMEOUT_FACTOR = 18;
 
     private final SyncStatusTracker initialSyncStatusTracker;


@@ -62,16 +60,18 @@ public class Follower extends AbstractRaftActorBehavior {
     private String leaderId;
     private short leaderPayloadVersion;
 
     private String leaderId;
     private short leaderPayloadVersion;
 

-    public Follower(RaftActorContext context) {
+    public Follower(final RaftActorContext context) {

         this(context, null, (short)-1);
     }
 
         this(context, null, (short)-1);
     }
 

-    public Follower(RaftActorContext context, String initialLeaderId, short initialLeaderPayloadVersion) {
+    public Follower(final RaftActorContext context, final String initialLeaderId,
+            final short initialLeaderPayloadVersion) {

         super(context, RaftState.Follower);
         this.leaderId = initialLeaderId;
         this.leaderPayloadVersion = initialLeaderPayloadVersion;
 
         super(context, RaftState.Follower);
         this.leaderId = initialLeaderId;
         this.leaderPayloadVersion = initialLeaderPayloadVersion;
 

-        initialSyncStatusTracker = new SyncStatusTracker(context.getActor(), getId(), SYNC_THRESHOLD);
+        initialSyncStatusTracker = new SyncStatusTracker(context.getActor(), getId(), context.getConfigParams()
+            .getSyncIndexThreshold());

 
         if (context.getPeerIds().isEmpty() && getLeaderId() == null) {
             actor().tell(TimeoutNow.INSTANCE, actor());
 
         if (context.getPeerIds().isEmpty() && getLeaderId() == null) {
             actor().tell(TimeoutNow.INSTANCE, actor());


@@ -96,7 +96,7 @@ public class Follower extends AbstractRaftActorBehavior {
     }
 
     @VisibleForTesting
     }
 
     @VisibleForTesting

-    protected final void setLeaderPayloadVersion(short leaderPayloadVersion) {
+    protected final void setLeaderPayloadVersion(final short leaderPayloadVersion) {

         this.leaderPayloadVersion = leaderPayloadVersion;
     }
 
         this.leaderPayloadVersion = leaderPayloadVersion;
     }
 


@@ -108,7 +108,7 @@ public class Follower extends AbstractRaftActorBehavior {
         lastLeaderMessageTimer.start();
     }
 
         lastLeaderMessageTimer.start();
     }
 

-    private boolean isLogEntryPresent(long index) {
+    private boolean isLogEntryPresent(final long index) {

         if (context.getReplicatedLog().isInSnapshot(index)) {
             return true;
         }
         if (context.getReplicatedLog().isInSnapshot(index)) {
             return true;
         }


@@ -118,12 +118,12 @@ public class Follower extends AbstractRaftActorBehavior {
 
     }
 
 
     }
 

-    private void updateInitialSyncStatus(long currentLeaderCommit, String newLeaderId) {
+    private void updateInitialSyncStatus(final long currentLeaderCommit, final String newLeaderId) {

         initialSyncStatusTracker.update(newLeaderId, currentLeaderCommit, context.getCommitIndex());
     }
 
     @Override
         initialSyncStatusTracker.update(newLeaderId, currentLeaderCommit, context.getCommitIndex());
     }
 
     @Override

-    protected RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries) {
+    protected RaftActorBehavior handleAppendEntries(final ActorRef sender, final AppendEntries appendEntries) {

 
         int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
         if (log.isTraceEnabled()) {
 
         int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
         if (log.isTraceEnabled()) {


@@ -317,7 +317,7 @@ public class Follower extends AbstractRaftActorBehavior {
         return this;
     }
 
         return this;
     }
 

-    private boolean isOutOfSync(AppendEntries appendEntries) {
+    private boolean isOutOfSync(final AppendEntries appendEntries) {

 
         long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex());
         boolean prevEntryPresent = isLogEntryPresent(appendEntries.getPrevLogIndex());
 
         long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex());
         boolean prevEntryPresent = isLogEntryPresent(appendEntries.getPrevLogIndex());


@@ -371,19 +371,19 @@ public class Follower extends AbstractRaftActorBehavior {
     }
 
     @Override
     }
 
     @Override

-    protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
-        AppendEntriesReply appendEntriesReply) {
+    protected RaftActorBehavior handleAppendEntriesReply(final ActorRef sender,
+        final AppendEntriesReply appendEntriesReply) {

         return this;
     }
 
     @Override
         return this;
     }
 
     @Override

-    protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
-        RequestVoteReply requestVoteReply) {
+    protected RaftActorBehavior handleRequestVoteReply(final ActorRef sender,
+        final RequestVoteReply requestVoteReply) {

         return this;
     }
 
     @Override
         return this;
     }
 
     @Override

-    public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
+    public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) {

         if (message instanceof ElectionTimeout || message instanceof TimeoutNow) {
             return handleElectionTimeout(message);
         }
         if (message instanceof ElectionTimeout || message instanceof TimeoutNow) {
             return handleElectionTimeout(message);
         }


@@ -419,7 +419,7 @@ public class Follower extends AbstractRaftActorBehavior {
         return super.handleMessage(sender, rpc);
     }
 
         return super.handleMessage(sender, rpc);
     }
 

-    private RaftActorBehavior handleElectionTimeout(Object message) {
+    private RaftActorBehavior handleElectionTimeout(final Object message) {

         // If the message is ElectionTimeout, verify we haven't actually seen a message from the leader
         // during the election timeout interval. It may that the election timer expired b/c this actor
         // was busy and messages got delayed, in which case leader messages would be backed up in the
         // If the message is ElectionTimeout, verify we haven't actually seen a message from the leader
         // during the election timeout interval. It may that the election timer expired b/c this actor
         // was busy and messages got delayed, in which case leader messages would be backed up in the


@@ -512,7 +512,7 @@ public class Follower extends AbstractRaftActorBehavior {
         return false;
     }
 
         return false;
     }
 

-    private void handleInstallSnapshot(final ActorRef sender, InstallSnapshot installSnapshot) {
+    private void handleInstallSnapshot(final ActorRef sender, final InstallSnapshot installSnapshot) {

 
         log.debug("{}: handleInstallSnapshot: {}", logName(), installSnapshot);
 
 
         log.debug("{}: handleInstallSnapshot: {}", logName(), installSnapshot);
 

diff --git a/opendaylight/md-sal/sal-clustering-config/src/main/resources/initial/datastore.cfg b/opendaylight/md-sal/sal-clustering-config/src/main/resources/initial/datastore.cfg
index 6d4d3b404ef9427086fcad04993358799bf234be..3ab714a9acd16f9a22ffebbeb63276fb73afc239 100644 (file)
--- a/opendaylight/md-sal/sal-clustering-config/src/main/resources/initial/datastore.cfg
+++ b/opendaylight/md-sal/sal-clustering-config/src/main/resources/initial/datastore.cfg
@@ -84,3 +84,7 @@ operational.persistent=false
 # Enable tell-based protocol between frontend (applications) and backend (shards). Using this protocol
 # should avoid AskTimeoutExceptions seen under heavy load. Defaults to false (use ask-based protocol).
 #use-tell-based-protocol=true
 # Enable tell-based protocol between frontend (applications) and backend (shards). Using this protocol
 # should avoid AskTimeoutExceptions seen under heavy load. Defaults to false (use ask-based protocol).
 #use-tell-based-protocol=true

+
+# Tune the maximum number of entries a follower is allowed to lag behind the leader before it is
+# considered out-of-sync. This flag may require tuning in face of a large number of small transactions.
+#sync-index-threshold=10

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java
index eeb6ad3155e2bc91d9e910728d589cca2084603b..02f2768fbb2a74f3fce79994800dbfb88da774d9 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DatastoreContext.java
@@ -11,8 +11,8 @@ package org.opendaylight.controller.cluster.datastore;
 import akka.util.Timeout;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import akka.util.Timeout;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;

-import com.google.common.collect.Sets;

 import java.util.Set;
 import java.util.Set;

+import java.util.concurrent.ConcurrentHashMap;

 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.text.WordUtils;
 import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.text.WordUtils;
 import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader;


@@ -56,7 +56,11 @@ public class DatastoreContext {
             TimeUnit.MILLISECONDS.convert(2, TimeUnit.MINUTES);
     public static final int DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE = 2048000;
 
             TimeUnit.MILLISECONDS.convert(2, TimeUnit.MINUTES);
     public static final int DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE = 2048000;
 

-    private static final Set<String> GLOBAL_DATASTORE_NAMES = Sets.newConcurrentHashSet();
+    public static final long DEFAULT_SYNC_INDEX_THRESHOLD = 10;
+
+    private static final Set<String> GLOBAL_DATASTORE_NAMES = ConcurrentHashMap.newKeySet();
+
+    private final DefaultConfigParamsImpl raftConfig = new DefaultConfigParamsImpl();

 
     private InMemoryDOMDataStoreConfigProperties dataStoreProperties;
     private Duration shardTransactionIdleTimeout = DatastoreContext.DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT;
 
     private InMemoryDOMDataStoreConfigProperties dataStoreProperties;
     private Duration shardTransactionIdleTimeout = DatastoreContext.DEFAULT_SHARD_TRANSACTION_IDLE_TIMEOUT;


@@ -69,7 +73,6 @@ public class DatastoreContext {
     private boolean persistent = DEFAULT_PERSISTENT;
     private AkkaConfigurationReader configurationReader = DEFAULT_CONFIGURATION_READER;
     private long transactionCreationInitialRateLimit = DEFAULT_TX_CREATION_INITIAL_RATE_LIMIT;
     private boolean persistent = DEFAULT_PERSISTENT;
     private AkkaConfigurationReader configurationReader = DEFAULT_CONFIGURATION_READER;
     private long transactionCreationInitialRateLimit = DEFAULT_TX_CREATION_INITIAL_RATE_LIMIT;

-    private final DefaultConfigParamsImpl raftConfig = new DefaultConfigParamsImpl();

     private String dataStoreName = UNKNOWN_DATA_STORE_TYPE;
     private LogicalDatastoreType logicalStoreType = LogicalDatastoreType.OPERATIONAL;
     private YangInstanceIdentifier storeRoot = YangInstanceIdentifier.EMPTY;
     private String dataStoreName = UNKNOWN_DATA_STORE_TYPE;
     private LogicalDatastoreType logicalStoreType = LogicalDatastoreType.OPERATIONAL;
     private YangInstanceIdentifier storeRoot = YangInstanceIdentifier.EMPTY;


@@ -92,6 +95,7 @@ public class DatastoreContext {
         setSnapshotDataThresholdPercentage(DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE);
         setElectionTimeoutFactor(DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR);
         setShardSnapshotChunkSize(DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE);
         setSnapshotDataThresholdPercentage(DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE);
         setElectionTimeoutFactor(DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR);
         setShardSnapshotChunkSize(DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE);

+        setSyncIndexThreshold(DEFAULT_SYNC_INDEX_THRESHOLD);

     }
 
     private DatastoreContext(final DatastoreContext other) {
     }
 
     private DatastoreContext(final DatastoreContext other) {


@@ -127,13 +131,14 @@ public class DatastoreContext {
         setPeerAddressResolver(other.raftConfig.getPeerAddressResolver());
         setTempFileDirectory(other.getTempFileDirectory());
         setFileBackedStreamingThreshold(other.getFileBackedStreamingThreshold());
         setPeerAddressResolver(other.raftConfig.getPeerAddressResolver());
         setTempFileDirectory(other.getTempFileDirectory());
         setFileBackedStreamingThreshold(other.getFileBackedStreamingThreshold());

+        setSyncIndexThreshold(other.raftConfig.getSyncIndexThreshold());

     }
 
     public static Builder newBuilder() {
         return new Builder(new DatastoreContext());
     }
 
     }
 
     public static Builder newBuilder() {
         return new Builder(new DatastoreContext());
     }
 

-    public static Builder newBuilderFrom(DatastoreContext context) {
+    public static Builder newBuilderFrom(final DatastoreContext context) {

         return new Builder(new DatastoreContext(context));
     }
 
         return new Builder(new DatastoreContext(context));
     }
 


@@ -209,7 +214,7 @@ public class DatastoreContext {
         return raftConfig.getTempFileDirectory();
     }
 
         return raftConfig.getTempFileDirectory();
     }
 

-    private void setTempFileDirectory(String tempFileDirectory) {
+    private void setTempFileDirectory(final String tempFileDirectory) {

         raftConfig.setTempFileDirectory(tempFileDirectory);
     }
 
         raftConfig.setTempFileDirectory(tempFileDirectory);
     }
 


@@ -217,51 +222,55 @@ public class DatastoreContext {
         return raftConfig.getFileBackedStreamingThreshold();
     }
 
         return raftConfig.getFileBackedStreamingThreshold();
     }
 

-    private void setFileBackedStreamingThreshold(int fileBackedStreamingThreshold) {
+    private void setFileBackedStreamingThreshold(final int fileBackedStreamingThreshold) {

         raftConfig.setFileBackedStreamingThreshold(fileBackedStreamingThreshold);
     }
 
         raftConfig.setFileBackedStreamingThreshold(fileBackedStreamingThreshold);
     }
 

-    private void setPeerAddressResolver(PeerAddressResolver resolver) {
+    private void setPeerAddressResolver(final PeerAddressResolver resolver) {

         raftConfig.setPeerAddressResolver(resolver);
     }
 
         raftConfig.setPeerAddressResolver(resolver);
     }
 

-    private void setHeartbeatInterval(long shardHeartbeatIntervalInMillis) {
+    private void setHeartbeatInterval(final long shardHeartbeatIntervalInMillis) {

         raftConfig.setHeartBeatInterval(new FiniteDuration(shardHeartbeatIntervalInMillis,
                 TimeUnit.MILLISECONDS));
     }
 
         raftConfig.setHeartBeatInterval(new FiniteDuration(shardHeartbeatIntervalInMillis,
                 TimeUnit.MILLISECONDS));
     }
 

-    private void setShardJournalRecoveryLogBatchSize(int shardJournalRecoveryLogBatchSize) {
+    private void setShardJournalRecoveryLogBatchSize(final int shardJournalRecoveryLogBatchSize) {

         raftConfig.setJournalRecoveryLogBatchSize(shardJournalRecoveryLogBatchSize);
     }
 
 
         raftConfig.setJournalRecoveryLogBatchSize(shardJournalRecoveryLogBatchSize);
     }
 
 

-    private void setIsolatedLeaderCheckInterval(long shardIsolatedLeaderCheckIntervalInMillis) {
+    private void setIsolatedLeaderCheckInterval(final long shardIsolatedLeaderCheckIntervalInMillis) {

         raftConfig.setIsolatedLeaderCheckInterval(
                 new FiniteDuration(shardIsolatedLeaderCheckIntervalInMillis, TimeUnit.MILLISECONDS));
     }
 
         raftConfig.setIsolatedLeaderCheckInterval(
                 new FiniteDuration(shardIsolatedLeaderCheckIntervalInMillis, TimeUnit.MILLISECONDS));
     }
 

-    private void setElectionTimeoutFactor(long shardElectionTimeoutFactor) {
+    private void setElectionTimeoutFactor(final long shardElectionTimeoutFactor) {

         raftConfig.setElectionTimeoutFactor(shardElectionTimeoutFactor);
     }
 
         raftConfig.setElectionTimeoutFactor(shardElectionTimeoutFactor);
     }
 

-    private void setCustomRaftPolicyImplementation(String customRaftPolicyImplementation) {
+    private void setCustomRaftPolicyImplementation(final String customRaftPolicyImplementation) {

         raftConfig.setCustomRaftPolicyImplementationClass(customRaftPolicyImplementation);
     }
 
         raftConfig.setCustomRaftPolicyImplementationClass(customRaftPolicyImplementation);
     }
 

-    private void setSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) {
+    private void setSnapshotDataThresholdPercentage(final int shardSnapshotDataThresholdPercentage) {

         Preconditions.checkArgument(shardSnapshotDataThresholdPercentage >= 0
                 && shardSnapshotDataThresholdPercentage <= 100);
         raftConfig.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
     }
 
         Preconditions.checkArgument(shardSnapshotDataThresholdPercentage >= 0
                 && shardSnapshotDataThresholdPercentage <= 100);
         raftConfig.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
     }
 

-    private void setSnapshotBatchCount(long shardSnapshotBatchCount) {
+    private void setSnapshotBatchCount(final long shardSnapshotBatchCount) {

         raftConfig.setSnapshotBatchCount(shardSnapshotBatchCount);
     }
 
         raftConfig.setSnapshotBatchCount(shardSnapshotBatchCount);
     }
 

-    private void setShardSnapshotChunkSize(int shardSnapshotChunkSize) {
+    private void setShardSnapshotChunkSize(final int shardSnapshotChunkSize) {

         raftConfig.setSnapshotChunkSize(shardSnapshotChunkSize);
     }
 
         raftConfig.setSnapshotChunkSize(shardSnapshotChunkSize);
     }
 

+    private void setSyncIndexThreshold(final long syncIndexThreshold) {
+        raftConfig.setSyncIndexThreshold(syncIndexThreshold);
+    }
+

     public int getShardBatchedModificationCount() {
         return shardBatchedModificationCount;
     }
     public int getShardBatchedModificationCount() {
         return shardBatchedModificationCount;
     }


@@ -286,7 +295,7 @@ public class DatastoreContext {
         return raftConfig.getSnapshotChunkSize();
     }
 
         return raftConfig.getSnapshotChunkSize();
     }
 

-    public static class Builder {
+    public static class Builder implements org.opendaylight.yangtools.concepts.Builder<DatastoreContext> {

         private final DatastoreContext datastoreContext;
         private int maxShardDataChangeExecutorPoolSize =
                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_EXECUTOR_POOL_SIZE;
         private final DatastoreContext datastoreContext;
         private int maxShardDataChangeExecutorPoolSize =
                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_EXECUTOR_POOL_SIZE;


@@ -297,7 +306,7 @@ public class DatastoreContext {
         private int maxShardDataStoreExecutorQueueSize =
                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_STORE_EXECUTOR_QUEUE_SIZE;
 
         private int maxShardDataStoreExecutorQueueSize =
                 InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_STORE_EXECUTOR_QUEUE_SIZE;
 

-        private Builder(DatastoreContext datastoreContext) {
+        private Builder(final DatastoreContext datastoreContext) {

             this.datastoreContext = datastoreContext;
 
             if (datastoreContext.getDataStoreProperties() != null) {
             this.datastoreContext = datastoreContext;
 
             if (datastoreContext.getDataStoreProperties() != null) {


@@ -312,115 +321,115 @@ public class DatastoreContext {
             }
         }
 
             }
         }
 

-        public Builder boundedMailboxCapacity(int boundedMailboxCapacity) {
+        public Builder boundedMailboxCapacity(final int boundedMailboxCapacity) {

             // TODO - this is defined in the yang DataStoreProperties but not currently used.
             return this;
         }
 
             // TODO - this is defined in the yang DataStoreProperties but not currently used.
             return this;
         }
 

-        public Builder enableMetricCapture(boolean enableMetricCapture) {
+        public Builder enableMetricCapture(final boolean enableMetricCapture) {

             // TODO - this is defined in the yang DataStoreProperties but not currently used.
             return this;
         }
 
 
             // TODO - this is defined in the yang DataStoreProperties but not currently used.
             return this;
         }
 
 

-        public Builder shardTransactionIdleTimeout(long timeout, TimeUnit unit) {
+        public Builder shardTransactionIdleTimeout(final long timeout, final TimeUnit unit) {

             datastoreContext.shardTransactionIdleTimeout = Duration.create(timeout, unit);
             return this;
         }
 
             datastoreContext.shardTransactionIdleTimeout = Duration.create(timeout, unit);
             return this;
         }
 

-        public Builder shardTransactionIdleTimeoutInMinutes(long timeout) {
+        public Builder shardTransactionIdleTimeoutInMinutes(final long timeout) {

             return shardTransactionIdleTimeout(timeout, TimeUnit.MINUTES);
         }
 
             return shardTransactionIdleTimeout(timeout, TimeUnit.MINUTES);
         }
 

-        public Builder operationTimeoutInSeconds(int operationTimeoutInSeconds) {
+        public Builder operationTimeoutInSeconds(final int operationTimeoutInSeconds) {

             datastoreContext.operationTimeoutInMillis = TimeUnit.SECONDS.toMillis(operationTimeoutInSeconds);
             return this;
         }
 
             datastoreContext.operationTimeoutInMillis = TimeUnit.SECONDS.toMillis(operationTimeoutInSeconds);
             return this;
         }
 

-        public Builder operationTimeoutInMillis(long operationTimeoutInMillis) {
+        public Builder operationTimeoutInMillis(final long operationTimeoutInMillis) {

             datastoreContext.operationTimeoutInMillis = operationTimeoutInMillis;
             return this;
         }
 
             datastoreContext.operationTimeoutInMillis = operationTimeoutInMillis;
             return this;
         }
 

-        public Builder dataStoreMXBeanType(String dataStoreMXBeanType) {
+        public Builder dataStoreMXBeanType(final String dataStoreMXBeanType) {

             datastoreContext.dataStoreMXBeanType = dataStoreMXBeanType;
             return this;
         }
 
             datastoreContext.dataStoreMXBeanType = dataStoreMXBeanType;
             return this;
         }
 

-        public Builder shardTransactionCommitTimeoutInSeconds(int shardTransactionCommitTimeoutInSeconds) {
+        public Builder shardTransactionCommitTimeoutInSeconds(final int shardTransactionCommitTimeoutInSeconds) {

             datastoreContext.shardTransactionCommitTimeoutInSeconds = shardTransactionCommitTimeoutInSeconds;
             return this;
         }
 
             datastoreContext.shardTransactionCommitTimeoutInSeconds = shardTransactionCommitTimeoutInSeconds;
             return this;
         }
 

-        public Builder shardJournalRecoveryLogBatchSize(int shardJournalRecoveryLogBatchSize) {
+        public Builder shardJournalRecoveryLogBatchSize(final int shardJournalRecoveryLogBatchSize) {

             datastoreContext.setShardJournalRecoveryLogBatchSize(shardJournalRecoveryLogBatchSize);
             return this;
         }
 
             datastoreContext.setShardJournalRecoveryLogBatchSize(shardJournalRecoveryLogBatchSize);
             return this;
         }
 

-        public Builder shardSnapshotBatchCount(int shardSnapshotBatchCount) {
+        public Builder shardSnapshotBatchCount(final int shardSnapshotBatchCount) {

             datastoreContext.setSnapshotBatchCount(shardSnapshotBatchCount);
             return this;
         }
 
             datastoreContext.setSnapshotBatchCount(shardSnapshotBatchCount);
             return this;
         }
 

-        public Builder shardSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) {
+        public Builder shardSnapshotDataThresholdPercentage(final int shardSnapshotDataThresholdPercentage) {

             datastoreContext.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
             return this;
         }
 
             datastoreContext.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
             return this;
         }
 

-        public Builder shardHeartbeatIntervalInMillis(int shardHeartbeatIntervalInMillis) {
+        public Builder shardHeartbeatIntervalInMillis(final int shardHeartbeatIntervalInMillis) {

             datastoreContext.setHeartbeatInterval(shardHeartbeatIntervalInMillis);
             return this;
         }
 
             datastoreContext.setHeartbeatInterval(shardHeartbeatIntervalInMillis);
             return this;
         }
 

-        public Builder shardTransactionCommitQueueCapacity(int shardTransactionCommitQueueCapacity) {
+        public Builder shardTransactionCommitQueueCapacity(final int shardTransactionCommitQueueCapacity) {

             datastoreContext.shardTransactionCommitQueueCapacity = shardTransactionCommitQueueCapacity;
             return this;
         }
 
             datastoreContext.shardTransactionCommitQueueCapacity = shardTransactionCommitQueueCapacity;
             return this;
         }
 

-        public Builder shardInitializationTimeout(long timeout, TimeUnit unit) {
+        public Builder shardInitializationTimeout(final long timeout, final TimeUnit unit) {

             datastoreContext.shardInitializationTimeout = new Timeout(timeout, unit);
             return this;
         }
 
             datastoreContext.shardInitializationTimeout = new Timeout(timeout, unit);
             return this;
         }
 

-        public Builder shardInitializationTimeoutInSeconds(long timeout) {
+        public Builder shardInitializationTimeoutInSeconds(final long timeout) {

             return shardInitializationTimeout(timeout, TimeUnit.SECONDS);
         }
 
             return shardInitializationTimeout(timeout, TimeUnit.SECONDS);
         }
 

-        public Builder shardLeaderElectionTimeout(long timeout, TimeUnit unit) {
+        public Builder shardLeaderElectionTimeout(final long timeout, final TimeUnit unit) {

             datastoreContext.shardLeaderElectionTimeout = new Timeout(timeout, unit);
             return this;
         }
 
             datastoreContext.shardLeaderElectionTimeout = new Timeout(timeout, unit);
             return this;
         }
 

-        public Builder shardLeaderElectionTimeoutInSeconds(long timeout) {
+        public Builder shardLeaderElectionTimeoutInSeconds(final long timeout) {

             return shardLeaderElectionTimeout(timeout, TimeUnit.SECONDS);
         }
 
             return shardLeaderElectionTimeout(timeout, TimeUnit.SECONDS);
         }
 

-        public Builder configurationReader(AkkaConfigurationReader configurationReader) {
+        public Builder configurationReader(final AkkaConfigurationReader configurationReader) {

             datastoreContext.configurationReader = configurationReader;
             return this;
         }
 
             datastoreContext.configurationReader = configurationReader;
             return this;
         }
 

-        public Builder persistent(boolean persistent) {
+        public Builder persistent(final boolean persistent) {

             datastoreContext.persistent = persistent;
             return this;
         }
 
             datastoreContext.persistent = persistent;
             return this;
         }
 

-        public Builder shardIsolatedLeaderCheckIntervalInMillis(int shardIsolatedLeaderCheckIntervalInMillis) {
+        public Builder shardIsolatedLeaderCheckIntervalInMillis(final int shardIsolatedLeaderCheckIntervalInMillis) {

             datastoreContext.setIsolatedLeaderCheckInterval(shardIsolatedLeaderCheckIntervalInMillis);
             return this;
         }
 
             datastoreContext.setIsolatedLeaderCheckInterval(shardIsolatedLeaderCheckIntervalInMillis);
             return this;
         }
 

-        public Builder shardElectionTimeoutFactor(long shardElectionTimeoutFactor) {
+        public Builder shardElectionTimeoutFactor(final long shardElectionTimeoutFactor) {

             datastoreContext.setElectionTimeoutFactor(shardElectionTimeoutFactor);
             return this;
         }
 
             datastoreContext.setElectionTimeoutFactor(shardElectionTimeoutFactor);
             return this;
         }
 

-        public Builder transactionCreationInitialRateLimit(long initialRateLimit) {
+        public Builder transactionCreationInitialRateLimit(final long initialRateLimit) {

             datastoreContext.transactionCreationInitialRateLimit = initialRateLimit;
             return this;
         }
 
             datastoreContext.transactionCreationInitialRateLimit = initialRateLimit;
             return this;
         }
 

-        public Builder logicalStoreType(LogicalDatastoreType logicalStoreType) {
+        public Builder logicalStoreType(final LogicalDatastoreType logicalStoreType) {

             datastoreContext.logicalStoreType = Preconditions.checkNotNull(logicalStoreType);
 
             // Retain compatible naming
             datastoreContext.logicalStoreType = Preconditions.checkNotNull(logicalStoreType);
 
             // Retain compatible naming


@@ -443,7 +452,7 @@ public class DatastoreContext {
             return this;
         }
 
             return this;
         }
 

-        public Builder dataStoreName(String dataStoreName) {
+        public Builder dataStoreName(final String dataStoreName) {

             datastoreContext.dataStoreName = Preconditions.checkNotNull(dataStoreName);
             datastoreContext.dataStoreMXBeanType = "Distributed" + WordUtils.capitalize(dataStoreName) + "Datastore";
             return this;
             datastoreContext.dataStoreName = Preconditions.checkNotNull(dataStoreName);
             datastoreContext.dataStoreMXBeanType = "Distributed" + WordUtils.capitalize(dataStoreName) + "Datastore";
             return this;


@@ -454,48 +463,48 @@ public class DatastoreContext {
             return this;
         }
 
             return this;
         }
 

-        public Builder writeOnlyTransactionOptimizationsEnabled(boolean value) {
+        public Builder writeOnlyTransactionOptimizationsEnabled(final boolean value) {

             datastoreContext.writeOnlyTransactionOptimizationsEnabled = value;
             return this;
         }
 
             datastoreContext.writeOnlyTransactionOptimizationsEnabled = value;
             return this;
         }
 

-        public Builder shardCommitQueueExpiryTimeoutInMillis(long value) {
+        public Builder shardCommitQueueExpiryTimeoutInMillis(final long value) {

             datastoreContext.shardCommitQueueExpiryTimeoutInMillis = value;
             return this;
         }
 
             datastoreContext.shardCommitQueueExpiryTimeoutInMillis = value;
             return this;
         }
 

-        public Builder shardCommitQueueExpiryTimeoutInSeconds(long value) {
+        public Builder shardCommitQueueExpiryTimeoutInSeconds(final long value) {

             datastoreContext.shardCommitQueueExpiryTimeoutInMillis = TimeUnit.MILLISECONDS.convert(
                     value, TimeUnit.SECONDS);
             return this;
         }
 
             datastoreContext.shardCommitQueueExpiryTimeoutInMillis = TimeUnit.MILLISECONDS.convert(
                     value, TimeUnit.SECONDS);
             return this;
         }
 

-        public Builder transactionDebugContextEnabled(boolean value) {
+        public Builder transactionDebugContextEnabled(final boolean value) {

             datastoreContext.transactionDebugContextEnabled = value;
             return this;
         }
 
             datastoreContext.transactionDebugContextEnabled = value;
             return this;
         }
 

-        public Builder maxShardDataChangeExecutorPoolSize(int maxShardDataChangeExecutorPoolSize) {
+        public Builder maxShardDataChangeExecutorPoolSize(final int maxShardDataChangeExecutorPoolSize) {

             this.maxShardDataChangeExecutorPoolSize = maxShardDataChangeExecutorPoolSize;
             return this;
         }
 
             this.maxShardDataChangeExecutorPoolSize = maxShardDataChangeExecutorPoolSize;
             return this;
         }
 

-        public Builder maxShardDataChangeExecutorQueueSize(int maxShardDataChangeExecutorQueueSize) {
+        public Builder maxShardDataChangeExecutorQueueSize(final int maxShardDataChangeExecutorQueueSize) {

             this.maxShardDataChangeExecutorQueueSize = maxShardDataChangeExecutorQueueSize;
             return this;
         }
 
             this.maxShardDataChangeExecutorQueueSize = maxShardDataChangeExecutorQueueSize;
             return this;
         }
 

-        public Builder maxShardDataChangeListenerQueueSize(int maxShardDataChangeListenerQueueSize) {
+        public Builder maxShardDataChangeListenerQueueSize(final int maxShardDataChangeListenerQueueSize) {

             this.maxShardDataChangeListenerQueueSize = maxShardDataChangeListenerQueueSize;
             return this;
         }
 
             this.maxShardDataChangeListenerQueueSize = maxShardDataChangeListenerQueueSize;
             return this;
         }
 

-        public Builder maxShardDataStoreExecutorQueueSize(int maxShardDataStoreExecutorQueueSize) {
+        public Builder maxShardDataStoreExecutorQueueSize(final int maxShardDataStoreExecutorQueueSize) {

             this.maxShardDataStoreExecutorQueueSize = maxShardDataStoreExecutorQueueSize;
             return this;
         }
 
             this.maxShardDataStoreExecutorQueueSize = maxShardDataStoreExecutorQueueSize;
             return this;
         }
 

-        public Builder useTellBasedProtocol(boolean value) {
+        public Builder useTellBasedProtocol(final boolean value) {

             datastoreContext.useTellBasedProtocol = value;
             return this;
         }
             datastoreContext.useTellBasedProtocol = value;
             return this;
         }


@@ -504,46 +513,52 @@ public class DatastoreContext {
          * For unit tests only.
          */
         @VisibleForTesting
          * For unit tests only.
          */
         @VisibleForTesting

-        public Builder shardManagerPersistenceId(String id) {
+        public Builder shardManagerPersistenceId(final String id) {

             datastoreContext.shardManagerPersistenceId = id;
             return this;
         }
 
             datastoreContext.shardManagerPersistenceId = id;
             return this;
         }
 

-        public DatastoreContext build() {
-            datastoreContext.dataStoreProperties = InMemoryDOMDataStoreConfigProperties.create(
-                    maxShardDataChangeExecutorPoolSize, maxShardDataChangeExecutorQueueSize,
-                    maxShardDataChangeListenerQueueSize, maxShardDataStoreExecutorQueueSize);
-
-            if (datastoreContext.dataStoreName != null) {
-                GLOBAL_DATASTORE_NAMES.add(datastoreContext.dataStoreName);
-            }
-
-            return datastoreContext;
-        }
-
-        public Builder customRaftPolicyImplementation(String customRaftPolicyImplementation) {
+        public Builder customRaftPolicyImplementation(final String customRaftPolicyImplementation) {

             datastoreContext.setCustomRaftPolicyImplementation(customRaftPolicyImplementation);
             return this;
         }
 
             datastoreContext.setCustomRaftPolicyImplementation(customRaftPolicyImplementation);
             return this;
         }
 

-        public Builder shardSnapshotChunkSize(int shardSnapshotChunkSize) {
+        public Builder shardSnapshotChunkSize(final int shardSnapshotChunkSize) {

             datastoreContext.setShardSnapshotChunkSize(shardSnapshotChunkSize);
             return this;
         }
 
             datastoreContext.setShardSnapshotChunkSize(shardSnapshotChunkSize);
             return this;
         }
 

-        public Builder shardPeerAddressResolver(PeerAddressResolver resolver) {
+        public Builder shardPeerAddressResolver(final PeerAddressResolver resolver) {

             datastoreContext.setPeerAddressResolver(resolver);
             return this;
         }
 
             datastoreContext.setPeerAddressResolver(resolver);
             return this;
         }
 

-        public Builder tempFileDirectory(String tempFileDirectory) {
+        public Builder tempFileDirectory(final String tempFileDirectory) {

             datastoreContext.setTempFileDirectory(tempFileDirectory);
             return this;
         }
 
             datastoreContext.setTempFileDirectory(tempFileDirectory);
             return this;
         }
 

-        public Builder fileBackedStreamingThresholdInMegabytes(int  fileBackedStreamingThreshold) {
+        public Builder fileBackedStreamingThresholdInMegabytes(final int  fileBackedStreamingThreshold) {

             datastoreContext.setFileBackedStreamingThreshold(fileBackedStreamingThreshold * ConfigParams.MEGABYTE);
             return this;
         }
             datastoreContext.setFileBackedStreamingThreshold(fileBackedStreamingThreshold * ConfigParams.MEGABYTE);
             return this;
         }

+
+        public Builder syncIndexThreshold(final long syncIndexThreshold) {
+            datastoreContext.setSyncIndexThreshold(syncIndexThreshold);
+            return this;
+        }
+
+        @Override
+        public DatastoreContext build() {
+            datastoreContext.dataStoreProperties = InMemoryDOMDataStoreConfigProperties.create(
+                    maxShardDataChangeExecutorPoolSize, maxShardDataChangeExecutorQueueSize,
+                    maxShardDataChangeListenerQueueSize, maxShardDataStoreExecutorQueueSize);
+
+            if (datastoreContext.dataStoreName != null) {
+                GLOBAL_DATASTORE_NAMES.add(datastoreContext.dataStoreName);
+            }
+
+            return datastoreContext;
+        }

     }
 }
     }
 }

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java
index 0847c981ee58567ed6178fe294c4545ef068d6ab..a99b8c3a986d8ef50dc46404c46e9b5a84c231bc 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java
@@ -20,13 +20,14 @@ public class DistributedConfigDataStoreProviderModule extends AbstractDistribute
     private BundleContext bundleContext;
 
     public DistributedConfigDataStoreProviderModule(
     private BundleContext bundleContext;
 
     public DistributedConfigDataStoreProviderModule(

-        org.opendaylight.controller.config.api.ModuleIdentifier identifier,
-        org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+        final org.opendaylight.controller.config.api.ModuleIdentifier identifier,
+        final org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {

         super(identifier, dependencyResolver);
     }
 
         super(identifier, dependencyResolver);
     }
 

-    public DistributedConfigDataStoreProviderModule(ModuleIdentifier identifier, DependencyResolver dependencyResolver,
-            DistributedConfigDataStoreProviderModule oldModule, AutoCloseable oldInstance) {
+    public DistributedConfigDataStoreProviderModule(final ModuleIdentifier identifier,
+            final DependencyResolver dependencyResolver, final DistributedConfigDataStoreProviderModule oldModule,
+            final AutoCloseable oldInstance) {

         super(identifier, dependencyResolver, oldModule, oldInstance);
     }
 
         super(identifier, dependencyResolver, oldModule, oldInstance);
     }
 


@@ -36,7 +37,7 @@ public class DistributedConfigDataStoreProviderModule extends AbstractDistribute
     }
 
     @Override
     }
 
     @Override

-    public boolean canReuseInstance(AbstractDistributedConfigDataStoreProviderModule oldModule) {
+    public boolean canReuseInstance(final AbstractDistributedConfigDataStoreProviderModule oldModule) {

         return true;
     }
 
         return true;
     }
 


@@ -54,7 +55,7 @@ public class DistributedConfigDataStoreProviderModule extends AbstractDistribute
         return newDatastoreContext(null);
     }
 
         return newDatastoreContext(null);
     }
 

-    private static DatastoreContext newDatastoreContext(ConfigProperties inProps) {
+    private static DatastoreContext newDatastoreContext(final ConfigProperties inProps) {

         ConfigProperties props = inProps;
         if (props == null) {
             props = new ConfigProperties();
         ConfigProperties props = inProps;
         if (props == null) {
             props = new ConfigProperties();


@@ -97,10 +98,11 @@ public class DistributedConfigDataStoreProviderModule extends AbstractDistribute
                 .customRaftPolicyImplementation(props.getCustomRaftPolicyImplementation())
                 .shardSnapshotChunkSize(props.getShardSnapshotChunkSize().getValue().intValue())
                 .useTellBasedProtocol(props.getUseTellBasedProtocol())
                 .customRaftPolicyImplementation(props.getCustomRaftPolicyImplementation())
                 .shardSnapshotChunkSize(props.getShardSnapshotChunkSize().getValue().intValue())
                 .useTellBasedProtocol(props.getUseTellBasedProtocol())

+                .syncIndexThreshold(props.getSyncIndexThreshold().getValue())

                 .build();
     }
 
                 .build();
     }
 

-    public void setBundleContext(BundleContext bundleContext) {
+    public void setBundleContext(final BundleContext bundleContext) {

         this.bundleContext = bundleContext;
     }
 }
         this.bundleContext = bundleContext;
     }
 }

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java
index 2e97c7ee9a7ba73ae4e2c3513ac953a9509f4173..44146aa8bb98a48567696d282386fc32075e053b 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java
@@ -20,14 +20,14 @@ public class DistributedOperationalDataStoreProviderModule
         extends AbstractDistributedOperationalDataStoreProviderModule {
     private BundleContext bundleContext;
 
         extends AbstractDistributedOperationalDataStoreProviderModule {
     private BundleContext bundleContext;
 

-    public DistributedOperationalDataStoreProviderModule(ModuleIdentifier identifier,
-            DependencyResolver dependencyResolver) {
+    public DistributedOperationalDataStoreProviderModule(final ModuleIdentifier identifier,
+            final DependencyResolver dependencyResolver) {

         super(identifier, dependencyResolver);
     }
 
         super(identifier, dependencyResolver);
     }
 

-    public DistributedOperationalDataStoreProviderModule(ModuleIdentifier identifier,
-            DependencyResolver dependencyResolver,DistributedOperationalDataStoreProviderModule oldModule,
-            AutoCloseable oldInstance) {
+    public DistributedOperationalDataStoreProviderModule(final ModuleIdentifier identifier,
+            final DependencyResolver dependencyResolver,final DistributedOperationalDataStoreProviderModule oldModule,
+            final AutoCloseable oldInstance) {

         super(identifier, dependencyResolver, oldModule, oldInstance);
     }
 
         super(identifier, dependencyResolver, oldModule, oldInstance);
     }
 


@@ -37,7 +37,7 @@ public class DistributedOperationalDataStoreProviderModule
     }
 
     @Override
     }
 
     @Override

-    public boolean canReuseInstance(AbstractDistributedOperationalDataStoreProviderModule oldModule) {
+    public boolean canReuseInstance(final AbstractDistributedOperationalDataStoreProviderModule oldModule) {

         return true;
     }
 
         return true;
     }
 


@@ -55,7 +55,7 @@ public class DistributedOperationalDataStoreProviderModule
         return newDatastoreContext(null);
     }
 
         return newDatastoreContext(null);
     }
 

-    private static DatastoreContext newDatastoreContext(OperationalProperties inProps) {
+    private static DatastoreContext newDatastoreContext(final OperationalProperties inProps) {

         OperationalProperties props = inProps;
         if (props == null) {
             props = new OperationalProperties();
         OperationalProperties props = inProps;
         if (props == null) {
             props = new OperationalProperties();


@@ -98,10 +98,11 @@ public class DistributedOperationalDataStoreProviderModule
                 .customRaftPolicyImplementation(props.getCustomRaftPolicyImplementation())
                 .shardSnapshotChunkSize(props.getShardSnapshotChunkSize().getValue().intValue())
                 .useTellBasedProtocol(props.getUseTellBasedProtocol())
                 .customRaftPolicyImplementation(props.getCustomRaftPolicyImplementation())
                 .shardSnapshotChunkSize(props.getShardSnapshotChunkSize().getValue().intValue())
                 .useTellBasedProtocol(props.getUseTellBasedProtocol())

+                .syncIndexThreshold(props.getSyncIndexThreshold().getValue())

                 .build();
     }
 
                 .build();
     }
 

-    public void setBundleContext(BundleContext bundleContext) {
+    public void setBundleContext(final BundleContext bundleContext) {

         this.bundleContext = bundleContext;
     }
 
         this.bundleContext = bundleContext;
     }
 

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/yang/distributed-datastore-provider.yang b/opendaylight/md-sal/sal-distributed-datastore/src/main/yang/distributed-datastore-provider.yang
index d14c828809e727eb8bf8dca08f1cae8cbc8f4d8e..49a51a625c9e7a4c9ce21f73c17c2c2e72d43357 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/yang/distributed-datastore-provider.yang
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/yang/distributed-datastore-provider.yang
@@ -248,6 +248,14 @@ module distributed-datastore-provider {
                 is the threshold in terms of number of megabytes before it should switch from storing in memory to
                 buffering to a file.";
         }
                 is the threshold in terms of number of megabytes before it should switch from storing in memory to
                 buffering to a file.";
         }

+
+        leaf sync-index-threshold {
+            default 10;
+            type non-zero-uint32-type;
+            description "Permitted synchronization lag, expressed in terms of RAFT entry count. It a follower's
+                         commitIndex trails the leader's journal by more than this amount of entries the follower
+                         is considered to be out-of-sync.";
+        }

     }
 
     // Augments the 'configuration' choice node under modules/module.
     }
 
     // Augments the 'configuration' choice node under modules/module.