You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by no...@apache.org on 2015/03/25 19:19:44 UTC
svn commit: r1669195 -
/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java
Author: noble
Date: Wed Mar 25 18:19:44 2015
New Revision: 1669195
URL: http://svn.apache.org/r1669195
Log:
SOLR-6924: conf node listening made more robust to take care of session expiry and reconnect
Modified:
lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java
Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java?rev=1669195&r1=1669194&r2=1669195&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/ZkController.java Wed Mar 25 18:19:44 2015
@@ -2297,8 +2297,7 @@ public final class ZkController {
log.info("watch zkdir " + zkDir);
if (!confDirectoryListeners.containsKey(zkDir)) {
confDirectoryListeners.put(zkDir, new HashSet<Runnable>());
- setConfWatcher(zkDir, new WatcherImpl(zkDir));
-
+ setConfWatcher(zkDir, new WatcherImpl(zkDir), null);
}
@@ -2312,46 +2311,66 @@ public final class ZkController {
@Override
public void process(WatchedEvent event) {
- try {
-
- synchronized (confDirectoryListeners) {
- // if this is not among directories to be watched then don't set the watcher anymore
- if( !confDirectoryListeners.containsKey(zkDir)) {
- log.info("Watcher on {} is removed ", zkDir);
- return;
- }
- Set<Runnable> listeners = confDirectoryListeners.get(zkDir);
- if (listeners != null && !listeners.isEmpty()) {
- final Set<Runnable> listenersCopy = new HashSet<>(listeners);
- new Thread() {
- //run these in a separate thread because this can be long running
- public void run() {
- for (final Runnable listener : listenersCopy) {
- try {
- listener.run();
- } catch (Exception e) {
- log.warn("listener throws error", e);
- }
- }
- }
- }.start();
- }
-
- }
+ Stat stat = null;
+ try {
+ stat = zkClient.exists(zkDir, null, true);
+ } catch (KeeperException e) {
+ //ignore , it is not a big deal
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ boolean resetWatcher = false;
+ try {
+ resetWatcher = fireEventListeners(zkDir);
} finally {
if (Event.EventType.None.equals(event.getType())) {
log.info("A node got unwatched for {}", zkDir);
- return;
} else {
- setConfWatcher(zkDir,this);
+ if (resetWatcher) setConfWatcher(zkDir, this, stat);
+ else log.info("A node got unwatched for {}", zkDir);
}
}
}
+
+ }
+
+ private boolean fireEventListeners(String zkDir) {
+ synchronized (confDirectoryListeners) {
+ // if this is not among directories to be watched then don't set the watcher anymore
+ if (!confDirectoryListeners.containsKey(zkDir)) {
+ log.info("Watcher on {} is removed ", zkDir);
+ return false;
+ }
+ Set<Runnable> listeners = confDirectoryListeners.get(zkDir);
+ if (listeners != null && !listeners.isEmpty()) {
+ final Set<Runnable> listenersCopy = new HashSet<>(listeners);
+ new Thread() {
+ //run these in a separate thread because this can be long running
+ public void run() {
+ for (final Runnable listener : listenersCopy) {
+ try {
+ listener.run();
+ } catch (Exception e) {
+ log.warn("listener throws error", e);
+ }
+ }
+ }
+ }.start();
+ }
+
}
+ return true;
+ }
- private void setConfWatcher(String zkDir, Watcher watcher) {
+ private void setConfWatcher(String zkDir, Watcher watcher, Stat stat) {
try {
+ Stat newStat = zkClient.exists(zkDir, watcher, true);
+ if (stat != null && newStat.getVersion() > stat.getVersion()) {
+ //a race condition where a we missed an even fired
+ //so fire the event listeners
+ fireEventListeners(zkDir);
+ }
zkClient.exists(zkDir, watcher, true);
} catch (KeeperException e) {
log.error("failed to set watcher for conf dir {} ", zkDir);
@@ -2368,6 +2387,7 @@ public final class ZkController {
synchronized (confDirectoryListeners){
for (String s : confDirectoryListeners.keySet()) {
watchZKConfDir(s);
+ fireEventListeners(s);
}
}
}