Added creationTime to Versioned object. This enables supporting a electedTime in leadership, which in turn helps us track how stable leadership terms are.
Change-Id: Ib051027625324646152ed85535ba337e95f8a061
diff --git a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastLeadershipService.java b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastLeadershipService.java
index b48ce21..d4b46ab 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastLeadershipService.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/HazelcastLeadershipService.java
@@ -170,7 +170,8 @@
if (topic != null) {
return new Leadership(topic.topicName(),
topic.leader(),
- topic.term());
+ topic.term(),
+ 0);
}
return null;
}
@@ -215,7 +216,8 @@
for (Topic topic : topics.values()) {
Leadership leadership = new Leadership(topic.topicName(),
topic.leader(),
- topic.term());
+ topic.term(),
+ 0);
result.put(topic.topicName(), leadership);
}
return result;
@@ -412,7 +414,7 @@
//
leadershipEvent = new LeadershipEvent(
LeadershipEvent.Type.LEADER_REELECTED,
- new Leadership(topicName, localNodeId, myLastLeaderTerm));
+ new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
// Dispatch to all instances
clusterCommunicator.broadcastIncludeSelf(
@@ -431,7 +433,7 @@
topicName, leader);
leadershipEvent = new LeadershipEvent(
LeadershipEvent.Type.LEADER_BOOTED,
- new Leadership(topicName, leader, myLastLeaderTerm));
+ new Leadership(topicName, leader, myLastLeaderTerm, 0));
// Dispatch only to the local listener(s)
eventDispatcher.post(leadershipEvent);
leader = null;
@@ -486,8 +488,8 @@
leader = localNodeId;
leadershipEvent = new LeadershipEvent(
- LeadershipEvent.Type.LEADER_ELECTED,
- new Leadership(topicName, localNodeId, myLastLeaderTerm));
+ LeadershipEvent.Type.LEADER_ELECTED,
+ new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
clusterCommunicator.broadcastIncludeSelf(
new ClusterMessage(
clusterService.getLocalNode().id(),
@@ -514,8 +516,8 @@
leader = null;
}
leadershipEvent = new LeadershipEvent(
- LeadershipEvent.Type.LEADER_BOOTED,
- new Leadership(topicName, localNodeId, myLastLeaderTerm));
+ LeadershipEvent.Type.LEADER_BOOTED,
+ new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
clusterCommunicator.broadcastIncludeSelf(
new ClusterMessage(
clusterService.getLocalNode().id(),
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/ConsistentMapImpl.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/ConsistentMapImpl.java
index 83eea18..0ceb566 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/ConsistentMapImpl.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/ConsistentMapImpl.java
@@ -105,7 +105,13 @@
public Versioned<V> get(K key) {
checkNotNull(key, ERROR_NULL_KEY);
Versioned<byte[]> value = complete(proxy.get(name, keyCache.getUnchecked(key)));
- return (value != null) ? new Versioned<>(serializer.decode(value.value()), value.version()) : null;
+ if (value == null) {
+ return null;
+ }
+ return new Versioned<>(
+ serializer.decode(value.value()),
+ value.version(),
+ value.creationTime());
}
@Override
@@ -114,16 +120,26 @@
checkNotNull(value, ERROR_NULL_VALUE);
Versioned<byte[]> previousValue =
complete(proxy.put(name, keyCache.getUnchecked(key), serializer.encode(value)));
- return (previousValue != null) ?
- new Versioned<>(serializer.decode(previousValue.value()), previousValue.version()) : null;
-
+ if (previousValue == null) {
+ return null;
+ }
+ return new Versioned<>(
+ serializer.decode(previousValue.value()),
+ previousValue.version(),
+ previousValue.creationTime());
}
@Override
public Versioned<V> remove(K key) {
checkNotNull(key, ERROR_NULL_KEY);
Versioned<byte[]> value = complete(proxy.remove(name, keyCache.getUnchecked(key)));
- return (value != null) ? new Versioned<>(serializer.decode(value.value()), value.version()) : null;
+ if (value == null) {
+ return null;
+ }
+ return new Versioned<>(
+ serializer.decode(value.value()),
+ value.version(),
+ value.creationTime());
}
@Override
@@ -143,7 +159,7 @@
public Collection<Versioned<V>> values() {
return Collections.unmodifiableList(complete(proxy.values(name))
.stream()
- .map(v -> new Versioned<V>(serializer.decode(v.value()), v.version()))
+ .map(v -> new Versioned<V>(serializer.decode(v.value()), v.version(), v.creationTime()))
.collect(Collectors.toList()));
}
@@ -161,8 +177,13 @@
checkNotNull(value, ERROR_NULL_VALUE);
Versioned<byte[]> existingValue = complete(proxy.putIfAbsent(
name, keyCache.getUnchecked(key), serializer.encode(value)));
- return (existingValue != null) ?
- new Versioned<>(serializer.decode(existingValue.value()), existingValue.version()) : null;
+ if (existingValue == null) {
+ return null;
+ }
+ return new Versioned<>(
+ serializer.decode(existingValue.value()),
+ existingValue.version(),
+ existingValue.creationTime());
}
@Override
@@ -212,6 +233,7 @@
dK(e.getKey()),
new Versioned<>(
serializer.decode(e.getValue().value()),
- e.getValue().version()));
+ e.getValue().version(),
+ e.getValue().creationTime()));
}
}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DistributedLeadershipManager.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DistributedLeadershipManager.java
index 0d38dd3..5bbb4cb 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DistributedLeadershipManager.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DistributedLeadershipManager.java
@@ -227,7 +227,7 @@
if (currentLeader != null) {
if (localNodeId.equals(currentLeader.value())) {
log.info("Already has leadership for {}", path);
- notifyNewLeader(path, localNodeId, currentLeader.version());
+ notifyNewLeader(path, localNodeId, currentLeader.version(), currentLeader.creationTime());
} else {
// someone else has leadership. will retry after sometime.
retry(path);
@@ -237,7 +237,7 @@
log.info("Assumed leadership for {}", path);
// do a get again to get the version (epoch)
Versioned<NodeId> newLeader = lockMap.get(path);
- notifyNewLeader(path, localNodeId, newLeader.version());
+ notifyNewLeader(path, localNodeId, newLeader.version(), newLeader.creationTime());
} else {
// someone beat us to it.
retry(path);
@@ -249,8 +249,8 @@
}
}
- private void notifyNewLeader(String path, NodeId leader, long epoch) {
- Leadership newLeadership = new Leadership(path, leader, epoch);
+ private void notifyNewLeader(String path, NodeId leader, long epoch, long electedTime) {
+ Leadership newLeadership = new Leadership(path, leader, epoch, electedTime);
boolean updatedLeader = false;
synchronized (leaderBoard) {
Leadership currentLeader = leaderBoard.get(path);
@@ -271,8 +271,8 @@
}
}
- private void notifyRemovedLeader(String path, NodeId leader, long epoch) {
- Leadership oldLeadership = new Leadership(path, leader, epoch);
+ private void notifyRemovedLeader(String path, NodeId leader, long epoch, long electedTime) {
+ Leadership oldLeadership = new Leadership(path, leader, epoch, electedTime);
boolean updatedLeader = false;
synchronized (leaderBoard) {
Leadership currentLeader = leaderBoard.get(path);
@@ -346,12 +346,13 @@
String path = entry.getKey();
NodeId nodeId = entry.getValue().value();
long epoch = entry.getValue().version();
+ long creationTime = entry.getValue().creationTime();
if (clusterService.getState(nodeId) == ControllerNode.State.INACTIVE) {
log.info("Lock for {} is held by {} which is currently inactive", path, nodeId);
try {
if (lockMap.remove(path, epoch)) {
log.info("Purged stale lock held by {} for {}", nodeId, path);
- notifyRemovedLeader(path, nodeId, epoch);
+ notifyRemovedLeader(path, nodeId, epoch, creationTime);
}
} catch (Exception e) {
log.warn("Failed to purge stale lock held by {} for {}", nodeId, path, e);
@@ -362,7 +363,7 @@
try {
if (lockMap.remove(path, epoch)) {
log.info("Purged stale lock held by {} for {}", nodeId, path);
- notifyRemovedLeader(path, nodeId, epoch);
+ notifyRemovedLeader(path, nodeId, epoch, creationTime);
}
} catch (Exception e) {
log.warn("Failed to purge stale lock held by {} for {}", nodeId, path, e);
diff --git a/core/store/trivial/src/main/java/org/onosproject/store/trivial/impl/SimpleLeadershipManager.java b/core/store/trivial/src/main/java/org/onosproject/store/trivial/impl/SimpleLeadershipManager.java
index d61772e..35654f2 100644
--- a/core/store/trivial/src/main/java/org/onosproject/store/trivial/impl/SimpleLeadershipManager.java
+++ b/core/store/trivial/src/main/java/org/onosproject/store/trivial/impl/SimpleLeadershipManager.java
@@ -61,7 +61,7 @@
@Override
public Leadership getLeadership(String path) {
checkArgument(path != null);
- return elections.get(path) ? new Leadership(path, clusterService.getLocalNode().id(), 0) : null;
+ return elections.get(path) ? new Leadership(path, clusterService.getLocalNode().id(), 0, 0) : null;
}
@Override
@@ -79,7 +79,7 @@
elections.put(path, true);
for (LeadershipEventListener listener : listeners) {
listener.event(new LeadershipEvent(Type.LEADER_ELECTED,
- new Leadership(path, clusterService.getLocalNode().id(), 0)));
+ new Leadership(path, clusterService.getLocalNode().id(), 0, 0)));
}
}
@@ -88,7 +88,7 @@
elections.remove(path);
for (LeadershipEventListener listener : listeners) {
listener.event(new LeadershipEvent(Type.LEADER_BOOTED,
- new Leadership(path, clusterService.getLocalNode().id(), 0)));
+ new Leadership(path, clusterService.getLocalNode().id(), 0, 0)));
}
}