blob: 0b2b4cb0e84bededaf68215796122ed48a40841a [file] [log] [blame]
Madan Jampanid14166a2015-02-24 17:37:51 -08001package org.onosproject.store.consistent.impl;
2
Ayaka Koshibec19b8b82015-04-08 15:18:24 -07003import com.google.common.collect.ImmutableList;
Jonathan Harte649c752015-03-03 18:04:25 -08004import com.google.common.collect.ImmutableMap;
Ayaka Koshibec19b8b82015-04-08 15:18:24 -07005import com.google.common.collect.Lists;
Jonathan Harte649c752015-03-03 18:04:25 -08006import com.google.common.collect.Maps;
7import com.google.common.collect.Sets;
Ayaka Koshibec19b8b82015-04-08 15:18:24 -07008
Madan Jampanid14166a2015-02-24 17:37:51 -08009import org.apache.felix.scr.annotations.Activate;
10import org.apache.felix.scr.annotations.Component;
11import org.apache.felix.scr.annotations.Deactivate;
12import org.apache.felix.scr.annotations.Reference;
13import org.apache.felix.scr.annotations.ReferenceCardinality;
14import org.apache.felix.scr.annotations.Service;
Ayaka Koshibefd26a302015-04-13 13:59:54 -070015import org.apache.commons.lang3.mutable.MutableBoolean;
Madan Jampanid14166a2015-02-24 17:37:51 -080016import org.onlab.util.KryoNamespace;
17import org.onosproject.cluster.ClusterService;
Madan Jampanid14166a2015-02-24 17:37:51 -080018import org.onosproject.cluster.Leadership;
19import org.onosproject.cluster.LeadershipEvent;
20import org.onosproject.cluster.LeadershipEventListener;
21import org.onosproject.cluster.LeadershipService;
22import org.onosproject.cluster.NodeId;
23import org.onosproject.event.AbstractListenerRegistry;
24import org.onosproject.event.EventDeliveryService;
25import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
26import org.onosproject.store.cluster.messaging.ClusterMessage;
27import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
28import org.onosproject.store.cluster.messaging.MessageSubject;
29import org.onosproject.store.serializers.KryoNamespaces;
Madan Jampanid14166a2015-02-24 17:37:51 -080030import org.onosproject.store.service.ConsistentMap;
Ayaka Koshibec19b8b82015-04-08 15:18:24 -070031import org.onosproject.store.service.ConsistentMapException;
Madan Jampanid14166a2015-02-24 17:37:51 -080032import org.onosproject.store.service.Serializer;
33import org.onosproject.store.service.StorageService;
34import org.onosproject.store.service.Versioned;
35import org.slf4j.Logger;
36
Jonathan Harte649c752015-03-03 18:04:25 -080037import java.util.Map;
38import java.util.Map.Entry;
39import java.util.Objects;
40import java.util.Set;
Ayaka Koshibec19b8b82015-04-08 15:18:24 -070041import java.util.List;
Jonathan Harte649c752015-03-03 18:04:25 -080042import java.util.concurrent.ExecutorService;
43import java.util.concurrent.Executors;
44import java.util.concurrent.ScheduledExecutorService;
45import java.util.concurrent.TimeUnit;
46import java.util.stream.Collectors;
47
48import static com.google.common.base.Preconditions.checkArgument;
49import static org.onlab.util.Tools.groupedThreads;
50import static org.slf4j.LoggerFactory.getLogger;
Madan Jampanid14166a2015-02-24 17:37:51 -080051
Ayaka Koshibe4a3c2392015-04-22 18:03:13 -070052import static org.onosproject.cluster.ControllerNode.State.ACTIVE;
53import static org.onosproject.cluster.ControllerNode.State.INACTIVE;
54
Madan Jampanid14166a2015-02-24 17:37:51 -080055/**
56 * Distributed Lock Manager implemented on top of ConsistentMap.
57 * <p>
58 * This implementation makes use of cluster manager's failure
59 * detection capabilities to detect and purge stale locks.
60 * TODO: Ensure lock safety and liveness.
61 */
Madan Jampani7f72c3f2015-03-01 17:34:59 -080062@Component(immediate = true, enabled = true)
Madan Jampanid14166a2015-02-24 17:37:51 -080063@Service
64public class DistributedLeadershipManager implements LeadershipService {
65
66 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
67 protected StorageService storageService;
68
69 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
70 protected ClusterService clusterService;
71
72 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
73 protected ClusterCommunicationService clusterCommunicator;
74
75 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
76 protected EventDeliveryService eventDispatcher;
77
78 private static final MessageSubject LEADERSHIP_EVENT_MESSAGE_SUBJECT =
79 new MessageSubject("distributed-leadership-manager-events");
80
81 private final Logger log = getLogger(getClass());
82 private ExecutorService messageHandlingExecutor;
83 private ScheduledExecutorService retryLeaderLockExecutor;
84 private ScheduledExecutorService deadLockDetectionExecutor;
85 private ScheduledExecutorService leadershipStatusBroadcaster;
86
Ayaka Koshibec19b8b82015-04-08 15:18:24 -070087 private ConsistentMap<String, NodeId> leaderMap;
88 private ConsistentMap<String, List<NodeId>> candidateMap;
89
Madan Jampanid14166a2015-02-24 17:37:51 -080090 private AbstractListenerRegistry<LeadershipEvent, LeadershipEventListener>
91 listenerRegistry;
92 private final Map<String, Leadership> leaderBoard = Maps.newConcurrentMap();
Ayaka Koshibefd26a302015-04-13 13:59:54 -070093 private final Map<String, Leadership> candidateBoard = Maps.newConcurrentMap();
Madan Jampanid14166a2015-02-24 17:37:51 -080094 private NodeId localNodeId;
95
96 private Set<String> activeTopics = Sets.newConcurrentHashSet();
97
Ayaka Koshibec19b8b82015-04-08 15:18:24 -070098 private static final int ELECTION_JOIN_ATTEMPT_INTERVAL_SEC = 2;
Madan Jampanid14166a2015-02-24 17:37:51 -080099 private static final int DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC = 2;
100 private static final int DEADLOCK_DETECTION_INTERVAL_SEC = 2;
Madan Jampania14047d2015-02-25 12:23:02 -0800101 private static final int LEADERSHIP_STATUS_UPDATE_INTERVAL_SEC = 2;
Madan Jampanid14166a2015-02-24 17:37:51 -0800102
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700103 private static final int LEADER_CANDIDATE_POS = 0;
104
105 private static final Serializer SERIALIZER = Serializer.using(
106 new KryoNamespace.Builder().register(KryoNamespaces.API).build());
Madan Jampanid14166a2015-02-24 17:37:51 -0800107
108 @Activate
109 public void activate() {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700110 leaderMap = storageService.<String, NodeId>consistentMapBuilder()
111 .withName("onos-topic-leaders")
112 .withSerializer(SERIALIZER)
113 .withPartitionsDisabled().build();
114 candidateMap = storageService.<String, List<NodeId>>consistentMapBuilder()
115 .withName("onos-topic-candidates")
116 .withSerializer(SERIALIZER)
117 .withPartitionsDisabled().build();
Madan Jampanid14166a2015-02-24 17:37:51 -0800118
119 localNodeId = clusterService.getLocalNode().id();
120
121 messageHandlingExecutor = Executors.newSingleThreadExecutor(
122 groupedThreads("onos/store/leadership", "message-handler"));
123 retryLeaderLockExecutor = Executors.newScheduledThreadPool(
124 4, groupedThreads("onos/store/leadership", "election-thread-%d"));
125 deadLockDetectionExecutor = Executors.newSingleThreadScheduledExecutor(
126 groupedThreads("onos/store/leadership", "dead-lock-detector"));
127 leadershipStatusBroadcaster = Executors.newSingleThreadScheduledExecutor(
128 groupedThreads("onos/store/leadership", "peer-updater"));
129 clusterCommunicator.addSubscriber(
130 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
131 new InternalLeadershipEventListener(),
132 messageHandlingExecutor);
133
134 deadLockDetectionExecutor.scheduleWithFixedDelay(
135 this::purgeStaleLocks, 0, DEADLOCK_DETECTION_INTERVAL_SEC, TimeUnit.SECONDS);
136 leadershipStatusBroadcaster.scheduleWithFixedDelay(
Madan Jampania14047d2015-02-25 12:23:02 -0800137 this::sendLeadershipStatus, 0, LEADERSHIP_STATUS_UPDATE_INTERVAL_SEC, TimeUnit.SECONDS);
Madan Jampanid14166a2015-02-24 17:37:51 -0800138
139 listenerRegistry = new AbstractListenerRegistry<>();
140 eventDispatcher.addSink(LeadershipEvent.class, listenerRegistry);
141
142 log.info("Started.");
143 }
144
145 @Deactivate
146 public void deactivate() {
147 leaderBoard.forEach((topic, leadership) -> {
148 if (localNodeId.equals(leadership.leader())) {
149 withdraw(topic);
150 }
151 });
152
153 eventDispatcher.removeSink(LeadershipEvent.class);
154 clusterCommunicator.removeSubscriber(LEADERSHIP_EVENT_MESSAGE_SUBJECT);
155
156 messageHandlingExecutor.shutdown();
157 retryLeaderLockExecutor.shutdown();
158 deadLockDetectionExecutor.shutdown();
159 leadershipStatusBroadcaster.shutdown();
160
161 log.info("Stopped.");
162 }
163
164 @Override
165 public Map<String, Leadership> getLeaderBoard() {
166 return ImmutableMap.copyOf(leaderBoard);
167 }
168
169 @Override
Madan Jampanifd45d5e2015-04-20 13:33:21 -0700170 public Map<String, List<NodeId>> getCandidates() {
171 return Maps.toMap(candidateBoard.keySet(), this::getCandidates);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700172 }
173
174 @Override
175 public List<NodeId> getCandidates(String path) {
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700176 Leadership current = candidateBoard.get(path);
177 return current == null ? ImmutableList.of() : ImmutableList.copyOf(current.candidates());
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700178 }
179
180 @Override
Madan Jampanid14166a2015-02-24 17:37:51 -0800181 public NodeId getLeader(String path) {
182 Leadership leadership = leaderBoard.get(path);
183 return leadership != null ? leadership.leader() : null;
184 }
185
186 @Override
Madan Jampani59610512015-02-25 15:25:43 -0800187 public Leadership getLeadership(String path) {
188 checkArgument(path != null);
189 return leaderBoard.get(path);
190 }
191
192 @Override
193 public Set<String> ownedTopics(NodeId nodeId) {
194 checkArgument(nodeId != null);
195 return leaderBoard.entrySet()
196 .stream()
197 .filter(entry -> nodeId.equals(entry.getValue().leader()))
198 .map(Entry::getKey)
199 .collect(Collectors.toSet());
200 }
201
202 @Override
Madan Jampanid14166a2015-02-24 17:37:51 -0800203 public void runForLeadership(String path) {
Madan Jampani52860be2015-02-27 12:52:37 -0800204 log.debug("Running for leadership for topic: {}", path);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700205 try {
206 Versioned<List<NodeId>> candidates = candidateMap.get(path);
207 if (candidates != null) {
208 List<NodeId> candidateList = Lists.newArrayList(candidates.value());
209 if (!candidateList.contains(localNodeId)) {
210 candidateList.add(localNodeId);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700211 if (candidateMap.replace(path, candidates.version(), candidateList)) {
212 Versioned<List<NodeId>> newCandidates = candidateMap.get(path);
213 notifyCandidateAdded(
214 path, candidateList, newCandidates.version(), newCandidates.creationTime());
215 } else {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700216 rerunForLeadership(path);
217 return;
218 }
219 }
220 } else {
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700221 List<NodeId> candidateList = ImmutableList.of(localNodeId);
222 if ((candidateMap.putIfAbsent(path, candidateList) == null)) {
223 Versioned<List<NodeId>> newCandidates = candidateMap.get(path);
224 notifyCandidateAdded(path, candidateList, newCandidates.version(), newCandidates.creationTime());
225 } else {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700226 rerunForLeadership(path);
227 return;
228 }
229 }
230 log.debug("In the leadership race for topic {} with candidates {}", path, candidates);
231 activeTopics.add(path);
232 tryLeaderLock(path);
233 } catch (ConsistentMapException e) {
234 log.debug("Failed to enter topic leader race for {}. Retrying.", path, e);
235 rerunForLeadership(path);
236 }
Madan Jampanid14166a2015-02-24 17:37:51 -0800237 }
238
239 @Override
240 public void withdraw(String path) {
241 activeTopics.remove(path);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700242
Madan Jampanid14166a2015-02-24 17:37:51 -0800243 try {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700244 Versioned<NodeId> leader = leaderMap.get(path);
245 if (leader != null && Objects.equals(leader.value(), localNodeId)) {
246 if (leaderMap.remove(path, leader.version())) {
Jonathan Harte649c752015-03-03 18:04:25 -0800247 log.info("Gave up leadership for {}", path);
248 notifyRemovedLeader(path, localNodeId, leader.version(), leader.creationTime());
249 }
Madan Jampanid14166a2015-02-24 17:37:51 -0800250 }
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700251 // else we are not the current leader, can still be a candidate.
252 Versioned<List<NodeId>> candidates = candidateMap.get(path);
253 List<NodeId> candidateList = candidates != null
254 ? Lists.newArrayList(candidates.value())
255 : Lists.newArrayList();
256 if (!candidateList.remove(localNodeId)) {
257 return;
258 }
Ayaka Koshibe93451b42015-04-21 15:12:03 -0700259 if (candidateMap.replace(path, candidates.version(), candidateList)) {
Ayaka Koshibe67cf7de2015-04-17 11:47:27 -0700260 Versioned<List<NodeId>> newCandidates = candidateMap.get(path);
261 notifyCandidateRemoved(path, candidates.version(), candidates.creationTime(), newCandidates);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700262 } else {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700263 log.warn("Failed to withdraw from candidates list. Will retry");
264 retryWithdraw(path);
265 }
Madan Jampanid14166a2015-02-24 17:37:51 -0800266 } catch (Exception e) {
Madan Jampania14047d2015-02-25 12:23:02 -0800267 log.debug("Failed to verify (and clear) any lock this node might be holding for {}", path, e);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700268 retryWithdraw(path);
Madan Jampanid14166a2015-02-24 17:37:51 -0800269 }
270 }
271
272 @Override
273 public void addListener(LeadershipEventListener listener) {
274 listenerRegistry.addListener(listener);
275 }
276
277 @Override
278 public void removeListener(LeadershipEventListener listener) {
279 listenerRegistry.removeListener(listener);
280 }
281
282 private void tryLeaderLock(String path) {
283 if (!activeTopics.contains(path)) {
284 return;
285 }
Ayaka Koshibe0d886fc2015-04-23 11:53:41 -0700286 try {
287 Versioned<List<NodeId>> candidates = candidateMap.get(path);
288 if (candidates != null) {
289 List<NodeId> activeNodes = candidates.value().stream()
290 .filter(n -> clusterService.getState(n) == ACTIVE)
291 .collect(Collectors.toList());
292 if (localNodeId.equals(activeNodes.get(LEADER_CANDIDATE_POS))) {
293 leaderLockAttempt(path, candidates.value());
294 } else {
295 retryLock(path);
296 }
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700297 } else {
Ayaka Koshibe0d886fc2015-04-23 11:53:41 -0700298 throw new IllegalStateException("should not be here");
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700299 }
Ayaka Koshibe0d886fc2015-04-23 11:53:41 -0700300 } catch (Exception e) {
301 log.debug("Failed to fetch candidate information for {}", path, e);
302 retryLock(path);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700303 }
304 }
305
306 private void leaderLockAttempt(String path, List<NodeId> candidates) {
Madan Jampanid14166a2015-02-24 17:37:51 -0800307 try {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700308 Versioned<NodeId> currentLeader = leaderMap.get(path);
Madan Jampanid14166a2015-02-24 17:37:51 -0800309 if (currentLeader != null) {
310 if (localNodeId.equals(currentLeader.value())) {
311 log.info("Already has leadership for {}", path);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700312 // FIXME: candidates can get out of sync.
313 notifyNewLeader(
314 path, localNodeId, candidates, currentLeader.version(), currentLeader.creationTime());
Madan Jampanid14166a2015-02-24 17:37:51 -0800315 } else {
316 // someone else has leadership. will retry after sometime.
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700317 retryLock(path);
Madan Jampanid14166a2015-02-24 17:37:51 -0800318 }
319 } else {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700320 if (leaderMap.putIfAbsent(path, localNodeId) == null) {
Madan Jampanid14166a2015-02-24 17:37:51 -0800321 log.info("Assumed leadership for {}", path);
322 // do a get again to get the version (epoch)
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700323 Versioned<NodeId> newLeader = leaderMap.get(path);
324 // FIXME: candidates can get out of sync
325 notifyNewLeader(path, localNodeId, candidates, newLeader.version(), newLeader.creationTime());
Madan Jampanid14166a2015-02-24 17:37:51 -0800326 } else {
327 // someone beat us to it.
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700328 retryLock(path);
Madan Jampanid14166a2015-02-24 17:37:51 -0800329 }
330 }
331 } catch (Exception e) {
Madan Jampania14047d2015-02-25 12:23:02 -0800332 log.debug("Attempt to acquire leadership lock for topic {} failed", path, e);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700333 retryLock(path);
Madan Jampanid14166a2015-02-24 17:37:51 -0800334 }
335 }
336
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700337 private void notifyCandidateAdded(
338 String path, List<NodeId> candidates, long epoch, long electedTime) {
339 Leadership newInfo = new Leadership(path, candidates, epoch, electedTime);
340 final MutableBoolean updated = new MutableBoolean(false);
341 candidateBoard.compute(path, (k, current) -> {
342 if (current == null || current.epoch() < newInfo.epoch()) {
Ayaka Koshibe0d886fc2015-04-23 11:53:41 -0700343 log.debug("updating candidateboard with {}", newInfo);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700344 updated.setTrue();
345 return newInfo;
346 }
347 return current;
348 });
349 // maybe rethink types of candidates events
350 if (updated.booleanValue()) {
351 LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.CANDIDATES_CHANGED, newInfo);
352 notifyPeers(event);
353 }
354 }
355
356 private void notifyCandidateRemoved(
Ayaka Koshibe67cf7de2015-04-17 11:47:27 -0700357 String path, long oldEpoch, long oldTime, Versioned<List<NodeId>> candidates) {
358 Leadership newInfo = (candidates == null)
359 ? new Leadership(path, ImmutableList.of(), oldEpoch, oldTime)
360 : new Leadership(path, candidates.value(), candidates.version(), candidates.creationTime());
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700361 final MutableBoolean updated = new MutableBoolean(false);
Ayaka Koshibe67cf7de2015-04-17 11:47:27 -0700362
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700363 candidateBoard.compute(path, (k, current) -> {
Ayaka Koshibe93451b42015-04-21 15:12:03 -0700364 if (current != null && current.epoch() < newInfo.epoch()) {
365 updated.setTrue();
366 return newInfo;
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700367 }
368 return current;
369 });
370 // maybe rethink types of candidates events
371 if (updated.booleanValue()) {
Ayaka Koshibe67cf7de2015-04-17 11:47:27 -0700372 log.debug("updated candidateboard with removal: {}", newInfo);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700373 LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.CANDIDATES_CHANGED, newInfo);
374 notifyPeers(event);
375 }
376 }
377
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700378 private void notifyNewLeader(String path, NodeId leader,
379 List<NodeId> candidates, long epoch, long electedTime) {
380 Leadership newLeadership = new Leadership(path, leader, candidates, epoch, electedTime);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700381 final MutableBoolean updatedLeader = new MutableBoolean(false);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700382 log.debug("candidates for new Leadership {}", candidates);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700383 leaderBoard.compute(path, (k, currentLeader) -> {
Madan Jampanid14166a2015-02-24 17:37:51 -0800384 if (currentLeader == null || currentLeader.epoch() < epoch) {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700385 log.debug("updating leaderboard with new {}", newLeadership);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700386 updatedLeader.setTrue();
387 return newLeadership;
Madan Jampanid14166a2015-02-24 17:37:51 -0800388 }
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700389 return currentLeader;
390 });
Madan Jampanid14166a2015-02-24 17:37:51 -0800391
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700392 if (updatedLeader.booleanValue()) {
Madan Jampanid14166a2015-02-24 17:37:51 -0800393 LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, newLeadership);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700394 notifyPeers(event);
Madan Jampanid14166a2015-02-24 17:37:51 -0800395 }
396 }
397
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700398 private void notifyPeers(LeadershipEvent event) {
399 eventDispatcher.post(event);
Madan Jampani2bfa94c2015-04-11 05:03:49 -0700400 clusterCommunicator.broadcast(event,
401 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
402 SERIALIZER::encode);
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700403 }
404
Madan Jampani30a57f82015-03-02 12:19:41 -0800405 private void notifyRemovedLeader(String path, NodeId leader, long epoch, long electedTime) {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700406 Versioned<List<NodeId>> candidates = candidateMap.get(path);
407 Leadership oldLeadership = new Leadership(
408 path, leader, candidates.value(), epoch, electedTime);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700409 final MutableBoolean updatedLeader = new MutableBoolean(false);
410 leaderBoard.compute(path, (k, currentLeader) -> {
Madan Jampanid14166a2015-02-24 17:37:51 -0800411 if (currentLeader != null && currentLeader.epoch() == oldLeadership.epoch()) {
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700412 updatedLeader.setTrue();
413 return null;
Madan Jampanid14166a2015-02-24 17:37:51 -0800414 }
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700415 return currentLeader;
416 });
Madan Jampanid14166a2015-02-24 17:37:51 -0800417
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700418 if (updatedLeader.booleanValue()) {
Madan Jampanid14166a2015-02-24 17:37:51 -0800419 LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_BOOTED, oldLeadership);
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700420 notifyPeers(event);
Madan Jampanid14166a2015-02-24 17:37:51 -0800421 }
422 }
423
424 private class InternalLeadershipEventListener implements ClusterMessageHandler {
425
426 @Override
427 public void handle(ClusterMessage message) {
428 LeadershipEvent leadershipEvent =
429 SERIALIZER.decode(message.payload());
430
Jonathan Hartb59ac4e2015-03-24 18:00:34 -0700431 log.trace("Leadership Event: time = {} type = {} event = {}",
Madan Jampanid14166a2015-02-24 17:37:51 -0800432 leadershipEvent.time(), leadershipEvent.type(),
433 leadershipEvent);
434
435 Leadership leadershipUpdate = leadershipEvent.subject();
436 LeadershipEvent.Type eventType = leadershipEvent.type();
437 String topic = leadershipUpdate.topic();
438
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700439 MutableBoolean updateAccepted = new MutableBoolean(false);
440 if (eventType.equals(LeadershipEvent.Type.LEADER_ELECTED)) {
441 leaderBoard.compute(topic, (k, currentLeadership) -> {
Madan Jampanid14166a2015-02-24 17:37:51 -0800442 if (currentLeadership == null || currentLeadership.epoch() < leadershipUpdate.epoch()) {
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700443 updateAccepted.setTrue();
444 return leadershipUpdate;
Madan Jampanid14166a2015-02-24 17:37:51 -0800445 }
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700446 return currentLeadership;
447 });
448 } else if (eventType.equals(LeadershipEvent.Type.LEADER_BOOTED)) {
449 leaderBoard.compute(topic, (k, currentLeadership) -> {
Ayaka Koshibe941f8602015-04-15 14:17:08 -0700450 if (currentLeadership == null || currentLeadership.epoch() == leadershipUpdate.epoch()) {
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700451 updateAccepted.setTrue();
452 return null;
Madan Jampanid14166a2015-02-24 17:37:51 -0800453 }
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700454 return currentLeadership;
455 });
456 } else if (eventType.equals(LeadershipEvent.Type.CANDIDATES_CHANGED)) {
457 candidateBoard.compute(topic, (k, currentInfo) -> {
Ayaka Koshibe93451b42015-04-21 15:12:03 -0700458 if (currentInfo == null || currentInfo.epoch() < leadershipUpdate.epoch()) {
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700459 updateAccepted.setTrue();
460 return leadershipUpdate;
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700461 }
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700462 return currentInfo;
463 });
464 } else {
465 throw new IllegalStateException("Unknown event type.");
466 }
467
468 if (updateAccepted.booleanValue()) {
469 eventDispatcher.post(leadershipEvent);
Madan Jampanid14166a2015-02-24 17:37:51 -0800470 }
471 }
472 }
473
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700474 private void rerunForLeadership(String path) {
475 retryLeaderLockExecutor.schedule(
476 () -> runForLeadership(path),
477 ELECTION_JOIN_ATTEMPT_INTERVAL_SEC,
478 TimeUnit.SECONDS);
479 }
480
481 private void retryLock(String path) {
Madan Jampanid14166a2015-02-24 17:37:51 -0800482 retryLeaderLockExecutor.schedule(
483 () -> tryLeaderLock(path),
484 DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC,
485 TimeUnit.SECONDS);
486 }
487
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700488 private void retryWithdraw(String path) {
489 retryLeaderLockExecutor.schedule(
490 () -> withdraw(path),
491 DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC,
492 TimeUnit.SECONDS);
493 }
494
Madan Jampanid14166a2015-02-24 17:37:51 -0800495 private void purgeStaleLocks() {
496 try {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700497 leaderMap.entrySet()
498 .stream()
Ayaka Koshibe4a3c2392015-04-22 18:03:13 -0700499 .filter(e -> clusterService.getState(e.getValue().value()) == INACTIVE)
500 .filter(e -> activeTopics.contains(e.getKey()))
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700501 .forEach(entry -> {
502 String path = entry.getKey();
503 NodeId nodeId = entry.getValue().value();
504 long epoch = entry.getValue().version();
505 long creationTime = entry.getValue().creationTime();
Madan Jampanid14166a2015-02-24 17:37:51 -0800506 try {
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700507 if (leaderMap.remove(path, epoch)) {
Madan Jampania14047d2015-02-25 12:23:02 -0800508 log.info("Purged stale lock held by {} for {}", nodeId, path);
Madan Jampani30a57f82015-03-02 12:19:41 -0800509 notifyRemovedLeader(path, nodeId, epoch, creationTime);
Madan Jampanid14166a2015-02-24 17:37:51 -0800510 }
511 } catch (Exception e) {
512 log.warn("Failed to purge stale lock held by {} for {}", nodeId, path, e);
513 }
Ayaka Koshibec19b8b82015-04-08 15:18:24 -0700514 });
Madan Jampanid14166a2015-02-24 17:37:51 -0800515 } catch (Exception e) {
Madan Jampania14047d2015-02-25 12:23:02 -0800516 log.debug("Failed cleaning up stale locks", e);
Madan Jampanid14166a2015-02-24 17:37:51 -0800517 }
518 }
519
520 private void sendLeadershipStatus() {
Madan Jampania14047d2015-02-25 12:23:02 -0800521 try {
522 leaderBoard.forEach((path, leadership) -> {
523 if (leadership.leader().equals(localNodeId)) {
524 LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, leadership);
Madan Jampani2bfa94c2015-04-11 05:03:49 -0700525 clusterCommunicator.broadcast(event,
526 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
527 SERIALIZER::encode);
Madan Jampania14047d2015-02-25 12:23:02 -0800528 }
529 });
Ayaka Koshibefd26a302015-04-13 13:59:54 -0700530 candidateBoard.forEach((path, leadership) -> {
531 LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.CANDIDATES_CHANGED, leadership);
532 clusterCommunicator.broadcast(event,
533 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
534 SERIALIZER::encode);
535 });
Madan Jampania14047d2015-02-25 12:23:02 -0800536 } catch (Exception e) {
537 log.debug("Failed to send leadership updates", e);
538 }
Madan Jampanid14166a2015-02-24 17:37:51 -0800539 }
Madan Jampania14047d2015-02-25 12:23:02 -0800540}