package org.opendaylight.controller.cluster.datastore;
import akka.util.Timeout;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.common.actor.FileAkkaConfigurationReader;
import org.opendaylight.controller.cluster.raft.ConfigParams;
import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
+import org.opendaylight.controller.cluster.raft.PeerAddressResolver;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;
public static final String UNKNOWN_DATA_STORE_TYPE = "unknown";
public static final int DEFAULT_SHARD_BATCHED_MODIFICATION_COUNT = 1000;
public static final long DEFAULT_SHARD_COMMIT_QUEUE_EXPIRY_TIMEOUT_IN_MS = TimeUnit.MILLISECONDS.convert(2, TimeUnit.MINUTES);
+ public static final int DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE = 2048000;
private static Set<String> globalDatastoreTypes = Sets.newConcurrentHashSet();
private boolean writeOnlyTransactionOptimizationsEnabled = true;
private long shardCommitQueueExpiryTimeoutInMillis = DEFAULT_SHARD_COMMIT_QUEUE_EXPIRY_TIMEOUT_IN_MS;
private boolean transactionDebugContextEnabled = false;
- private String customRaftPolicyImplementation = "";
+ private String shardManagerPersistenceId;
public static Set<String> getGlobalDatastoreTypes() {
return globalDatastoreTypes;
setIsolatedLeaderCheckInterval(DEFAULT_ISOLATED_LEADER_CHECK_INTERVAL_IN_MILLIS);
setSnapshotDataThresholdPercentage(DEFAULT_SHARD_SNAPSHOT_DATA_THRESHOLD_PERCENTAGE);
setElectionTimeoutFactor(DEFAULT_SHARD_ELECTION_TIMEOUT_FACTOR);
+ setShardSnapshotChunkSize(DEFAULT_SHARD_SNAPSHOT_CHUNK_SIZE);
}
private DatastoreContext(DatastoreContext other) {
this.writeOnlyTransactionOptimizationsEnabled = other.writeOnlyTransactionOptimizationsEnabled;
this.shardCommitQueueExpiryTimeoutInMillis = other.shardCommitQueueExpiryTimeoutInMillis;
this.transactionDebugContextEnabled = other.transactionDebugContextEnabled;
- this.customRaftPolicyImplementation = other.customRaftPolicyImplementation;
+ this.shardManagerPersistenceId = shardManagerPersistenceId;
setShardJournalRecoveryLogBatchSize(other.raftConfig.getJournalRecoveryLogBatchSize());
setSnapshotBatchCount(other.raftConfig.getSnapshotBatchCount());
setIsolatedLeaderCheckInterval(other.raftConfig.getIsolatedCheckIntervalInMillis());
setSnapshotDataThresholdPercentage(other.raftConfig.getSnapshotDataThresholdPercentage());
setElectionTimeoutFactor(other.raftConfig.getElectionTimeoutFactor());
- setCustomRaftPolicyImplementation(other.customRaftPolicyImplementation);
-
+ setCustomRaftPolicyImplementation(other.raftConfig.getCustomRaftPolicyImplementationClass());
+ setShardSnapshotChunkSize(other.raftConfig.getSnapshotChunkSize());
+ setPeerAddressResolver(other.raftConfig.getPeerAddressResolver());
}
public static Builder newBuilder() {
return transactionCreationInitialRateLimit;
}
+ public String getShardManagerPersistenceId() {
+ return shardManagerPersistenceId;
+ }
+
+ private void setPeerAddressResolver(PeerAddressResolver resolver) {
+ raftConfig.setPeerAddressResolver(resolver);
+ }
+
private void setHeartbeatInterval(long shardHeartbeatIntervalInMillis){
raftConfig.setHeartBeatInterval(new FiniteDuration(shardHeartbeatIntervalInMillis,
TimeUnit.MILLISECONDS));
raftConfig.setCustomRaftPolicyImplementationClass(customRaftPolicyImplementation);
}
-
private void setSnapshotDataThresholdPercentage(int shardSnapshotDataThresholdPercentage) {
raftConfig.setSnapshotDataThresholdPercentage(shardSnapshotDataThresholdPercentage);
}
raftConfig.setSnapshotBatchCount(shardSnapshotBatchCount);
}
+ private void setShardSnapshotChunkSize(int shardSnapshotChunkSize) {
+ raftConfig.setSnapshotChunkSize(shardSnapshotChunkSize);
+ }
+
public int getShardBatchedModificationCount() {
return shardBatchedModificationCount;
}
return transactionDebugContextEnabled;
}
+ public int getShardSnapshotChunkSize() {
+ return raftConfig.getSnapshotChunkSize();
+ }
+
public static class Builder {
private final DatastoreContext datastoreContext;
private int maxShardDataChangeExecutorPoolSize =
return this;
}
+ /**
+ * For unit tests only.
+ */
+ @VisibleForTesting
+ public Builder shardManagerPersistenceId(String id) {
+ datastoreContext.shardManagerPersistenceId = id;
+ return this;
+ }
+
public DatastoreContext build() {
datastoreContext.dataStoreProperties = InMemoryDOMDataStoreConfigProperties.create(
maxShardDataChangeExecutorPoolSize, maxShardDataChangeExecutorQueueSize,
datastoreContext.setCustomRaftPolicyImplementation(customRaftPolicyImplementation);
return this;
}
+
+ public Builder shardSnapshotChunkSize(int shardSnapshotChunkSize) {
+ datastoreContext.setShardSnapshotChunkSize(shardSnapshotChunkSize);
+ return this;
+ }
+
+ public Builder shardPeerAddressResolver(PeerAddressResolver resolver) {
+ datastoreContext.setPeerAddressResolver(resolver);
+ return this;
+ }
}
}