Code Review
/
controller.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
BUG 2676 : Apply backpressure when creating transactions
[controller.git]
/
opendaylight
/
md-sal
/
sal-distributed-datastore
/
src
/
main
/
java
/
org
/
opendaylight
/
controller
/
cluster
/
datastore
/
ShardManager.java
diff --git
a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
index 88f818f0faedf76f0349ce1f7294dee37b9d79d1..9c8f0b24440775a68f6047e3be19210583e66dbb 100644
(file)
--- a/
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
+++ b/
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
@@
-15,8
+15,6
@@
import akka.actor.OneForOneStrategy;
import akka.actor.Props;
import akka.actor.SupervisorStrategy;
import akka.cluster.ClusterEvent;
import akka.actor.Props;
import akka.actor.SupervisorStrategy;
import akka.cluster.ClusterEvent;
-import akka.event.Logging;
-import akka.event.LoggingAdapter;
import akka.japi.Creator;
import akka.japi.Function;
import akka.japi.Procedure;
import akka.japi.Creator;
import akka.japi.Function;
import akka.japi.Procedure;
@@
-25,7
+23,17
@@
import akka.persistence.RecoveryFailure;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Lists;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
@@
-44,17
+52,10
@@
import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.Duration;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
/**
* The ShardManager has the following jobs,
* <ul>
/**
* The ShardManager has the following jobs,
* <ul>
@@
-66,8
+67,7
@@
import java.util.Set;
*/
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
*/
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
- protected final LoggingAdapter LOG =
- Logging.getLogger(getContext().system(), this);
+ private final Logger LOG = LoggerFactory.getLogger(getClass());
// Stores a mapping between a member name and the address of the member
// Member names look like "member-1", "member-2" etc and are as specified
// Stores a mapping between a member name and the address of the member
// Member names look like "member-1", "member-2" etc and are as specified
@@
-91,7
+91,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
private final DatastoreContext datastoreContext;
private final DatastoreContext datastoreContext;
- private
final Collection<String> knownModules = new HashSet<>(128
);
+ private
Collection<String> knownModules = Collections.emptySet(
);
private final DataPersistenceProvider dataPersistenceProvider;
private final DataPersistenceProvider dataPersistenceProvider;
@@
-182,11
+182,10
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
if(dataPersistenceProvider.isRecoveryApplicable()) {
if (message instanceof SchemaContextModules) {
SchemaContextModules msg = (SchemaContextModules) message;
if(dataPersistenceProvider.isRecoveryApplicable()) {
if (message instanceof SchemaContextModules) {
SchemaContextModules msg = (SchemaContextModules) message;
- knownModules.clear();
- knownModules.addAll(msg.getModules());
+ knownModules = ImmutableSet.copyOf(msg.getModules());
} else if (message instanceof RecoveryFailure) {
RecoveryFailure failure = (RecoveryFailure) message;
} else if (message instanceof RecoveryFailure) {
RecoveryFailure failure = (RecoveryFailure) message;
- LOG.error(
failure.cause(), "Recovery failed"
);
+ LOG.error(
"Recovery failed", failure.cause()
);
} else if (message instanceof RecoveryCompleted) {
LOG.info("Recovery complete : {}", persistenceId());
} else if (message instanceof RecoveryCompleted) {
LOG.info("Recovery complete : {}", persistenceId());
@@
-277,8
+276,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
LOG.info("New SchemaContext has a super set of current knownModules - persisting info");
LOG.info("New SchemaContext has a super set of current knownModules - persisting info");
- knownModules.clear();
- knownModules.addAll(newModules);
+ knownModules = ImmutableSet.copyOf(newModules);
dataPersistenceProvider.persist(new SchemaContextModules(newModules), new Procedure<SchemaContextModules>() {
dataPersistenceProvider.persist(new SchemaContextModules(newModules), new Procedure<SchemaContextModules>() {
@@
-425,12
+423,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
new Function<Throwable, SupervisorStrategy.Directive>() {
@Override
public SupervisorStrategy.Directive apply(Throwable t) {
new Function<Throwable, SupervisorStrategy.Directive>() {
@Override
public SupervisorStrategy.Directive apply(Throwable t) {
- StringBuilder sb = new StringBuilder();
- for(StackTraceElement element : t.getStackTrace()) {
- sb.append("\n\tat ")
- .append(element.toString());
- }
- LOG.warning("Supervisor Strategy of resume applied {}",sb.toString());
+ LOG.warn("Supervisor Strategy caught unexpected exception - resuming", t);
return SupervisorStrategy.resume();
}
}
return SupervisorStrategy.resume();
}
}
@@
-556,7
+549,8
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
}
static class SchemaContextModules implements Serializable {
}
static class SchemaContextModules implements Serializable {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = -8884620101025936590L;
+
private final Set<String> modules;
SchemaContextModules(Set<String> modules){
private final Set<String> modules;
SchemaContextModules(Set<String> modules){