Skip to content

Commit

Permalink
LocalNodeMasterListener is a regular listener (elastic#62422)
Browse files Browse the repository at this point in the history
This commit makes the LocalNodeMasterListener interface extend the
ClusterStateListener interface and use a default implementation for
detecting whether the local node master status changed.
  • Loading branch information
jaymode authored Sep 16, 2020
1 parent 86a0f15 commit fb4e4ff
Show file tree
Hide file tree
Showing 2 changed files with 13 additions and 45 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
* Enables listening to master changes events of the local node (when the local node becomes the master, and when the local
* node cease being a master).
*/
public interface LocalNodeMasterListener {
public interface LocalNodeMasterListener extends ClusterStateListener {

/**
* Called when local node is elected to be the master
Expand All @@ -33,5 +33,16 @@ public interface LocalNodeMasterListener {
* Called when the local node used to be the master, a new master was elected and it's no longer the local node.
*/
void offMaster();

@Override
default void clusterChanged(ClusterChangedEvent event) {
final boolean wasMaster = event.previousState().nodes().isLocalNodeElectedMaster();
final boolean isMaster = event.localNodeMaster();
if (wasMaster == false && isMaster) {
onMaster();
} else if (wasMaster && isMaster == false) {
offMaster();
}
}
}

Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,6 @@

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -92,8 +91,6 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
private final Collection<ClusterStateListener> clusterStateListeners = new CopyOnWriteArrayList<>();
private final Map<TimeoutClusterStateListener, NotifyTimeout> timeoutClusterStateListeners = new ConcurrentHashMap<>();

private final LocalNodeMasterListeners localNodeMasterListeners;

private final AtomicReference<ClusterState> state; // last applied state

private final String nodeName;
Expand All @@ -104,7 +101,6 @@ public ClusterApplierService(String nodeName, Settings settings, ClusterSettings
this.clusterSettings = clusterSettings;
this.threadPool = threadPool;
this.state = new AtomicReference<>();
this.localNodeMasterListeners = new LocalNodeMasterListeners();
this.nodeName = nodeName;

this.slowTaskLoggingThreshold = CLUSTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(settings);
Expand Down Expand Up @@ -134,7 +130,6 @@ public void setInitialState(ClusterState initialState) {
protected synchronized void doStart() {
Objects.requireNonNull(nodeConnectionsService, "please set the node connection service before starting");
Objects.requireNonNull(state.get(), "please set initial state before starting");
addListener(localNodeMasterListeners);
threadPoolExecutor = createThreadPoolExecutor();
}

Expand Down Expand Up @@ -179,7 +174,6 @@ protected synchronized void doStop() {
}
}
ThreadPool.terminate(threadPoolExecutor, 10, TimeUnit.SECONDS);
removeListener(localNodeMasterListeners);
}

@Override
Expand Down Expand Up @@ -255,7 +249,7 @@ public void removeTimeoutListener(TimeoutClusterStateListener listener) {
* Add a listener for on/off local node master events
*/
public void addLocalNodeMasterListener(LocalNodeMasterListener listener) {
localNodeMasterListeners.add(listener);
addListener(listener);
}

/**
Expand Down Expand Up @@ -600,43 +594,6 @@ public void run() {
}
}

private static class LocalNodeMasterListeners implements ClusterStateListener {

private final List<LocalNodeMasterListener> listeners = new CopyOnWriteArrayList<>();
private volatile boolean master = false;

private LocalNodeMasterListeners() {
}

@Override
public void clusterChanged(ClusterChangedEvent event) {
if (!master && event.localNodeMaster()) {
master = true;
for (LocalNodeMasterListener listener : listeners) {
try {
listener.onMaster();
} catch (Exception e) {
logger.warn("failed to notify LocalNodeMasterListener", e);
}
}
} else if (master && !event.localNodeMaster()) {
master = false;
for (LocalNodeMasterListener listener : listeners) {
try {
listener.offMaster();
} catch (Exception e) {
logger.warn("failed to notify LocalNodeMasterListener", e);
}
}
}
}

private void add(LocalNodeMasterListener listener) {
listeners.add(listener);
}

}

// this one is overridden in tests so we can control time
protected long currentTimeInMillis() {
return threadPool.relativeTimeInMillis();
Expand Down

0 comments on commit fb4e4ff

Please sign in to comment.