svn commit: r1233419 - /lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java

[ma...@apache.org]

Author: markrmiller
Date: Thu Jan 19 15:06:02 2012
New Revision: 1233419

URL: http://svn.apache.org/viewvc?rev=1233419&view=rev
Log:
some more logic about when to be leader

Modified:
    lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java

Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java?rev=1233419&r1=1233418&r2=1233419&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java Thu Jan 19 15:06:02 2012
@@ -130,6 +130,12 @@ final class ShardLeaderElectionContext e
   void runLeaderProcess(String leaderSeqPath, boolean weAreReplacement, SolrCore core)
       throws KeeperException, InterruptedException, IOException {
     
+    // should I be leader?
+    if (weAreReplacement && !shouldIBeLeader(leaderProps)) {
+      rejoinLeaderElection(leaderSeqPath, core);
+      return;
+    }
+    
     // TODO: move sync stuff to a better spot??
     if (weAreReplacement && core != null) { // TODO: core can be null in tests
       zkController.publish(core, ZkStateReader.SYNC);
@@ -141,15 +147,7 @@ final class ShardLeaderElectionContext e
       System.out.println("SYNC UP");
       boolean success = syncReplicas(core);
       if (!success) {
-        // remove our ephemeral and re join the election
-        System.out.println("sync failed, delete our election node:"
-            + leaderSeqPath);
-        zkController.publish(core, ZkStateReader.DOWN);
-        zkClient.delete(leaderSeqPath, -1, true);
-        
-        core.getUpdateHandler().getSolrCoreState().doRecovery(core);
-        
-        leaderElector.joinElection(this, core);
+        rejoinLeaderElection(leaderSeqPath, core);
         return;
       }
       
@@ -163,6 +161,19 @@ final class ShardLeaderElectionContext e
     
     super.runLeaderProcess(leaderSeqPath, weAreReplacement, core);
   }
+
+  private void rejoinLeaderElection(String leaderSeqPath, SolrCore core)
+      throws InterruptedException, KeeperException, IOException {
+    // remove our ephemeral and re join the election
+    System.out.println("sync failed, delete our election node:"
+        + leaderSeqPath);
+    zkController.publish(core, ZkStateReader.DOWN);
+    zkClient.delete(leaderSeqPath, -1, true);
+    
+    core.getUpdateHandler().getSolrCoreState().doRecovery(core);
+    
+    leaderElector.joinElection(this, core);
+  }
   
   private boolean syncReplicas(SolrCore core) {
     boolean success = false;
@@ -255,6 +266,38 @@ final class ShardLeaderElectionContext e
     return false;
   }
   
+  protected boolean shouldIBeLeader(ZkNodeProps leaderProps) {
+    CloudState cloudState = zkController.getZkStateReader().getCloudState();
+    Map<String,Slice> slices = cloudState.getSlices(this.collection);
+    Slice slice = slices.get(shardId);
+    Map<String,ZkNodeProps> shards = slice.getShards();
+    boolean foundSomeoneElseActive = false;
+    for (Map.Entry<String,ZkNodeProps> shard : shards.entrySet()) {
+      String state = shard.getValue().get(ZkStateReader.STATE_PROP);
+
+      if (new ZkCoreNodeProps(shard.getValue()).getCoreUrl().equals(
+              new ZkCoreNodeProps(leaderProps).getCoreUrl())) {
+        if (state.equals(ZkStateReader.ACTIVE)
+          && cloudState.liveNodesContain(shard.getValue().get(
+              ZkStateReader.NODE_NAME_PROP))) {
+          // we are alive
+          return true;
+        }
+      }
+      
+      if ((state.equals(ZkStateReader.ACTIVE))
+          && cloudState.liveNodesContain(shard.getValue().get(
+              ZkStateReader.NODE_NAME_PROP))
+          && !new ZkCoreNodeProps(shard.getValue()).getCoreUrl().equals(
+              new ZkCoreNodeProps(leaderProps).getCoreUrl())) {
+        foundSomeoneElseActive = true;
+      }
+    }
+    
+    System.out.println("FOUND ACTIVE: " + foundSomeoneElseActive);
+    return !foundSomeoneElseActive;
+  }
+  
   private boolean sync(SolrCore core, ZkNodeProps props) throws MalformedURLException,
       SolrServerException, IOException {
     List<ZkCoreNodeProps> nodes = zkController.getZkStateReader()