SOLR-4916: Add support to write and read Solr index files and transac…

…tion log files to and from HDFS.

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1497072 13f79535-47bb-0310-9956-ffa450edef68

.gitignore

@@ -1,4 +1,5 @@
-
+# hdfs
+/solr/example/hdfs
 *.jar
 
 # .

build.xml

@@ -205,7 +205,7 @@
     <!-- TODO: find a better way to exclude duplicate JAR files & fix the servlet-api mess! -->
     <pathconvert property="eclipse.fileset.libs" pathsep="|" dirsep="/">
       <fileset dir="${basedir}/lucene" includes="**/lib/*.jar" excludes="**/*servlet-api*.jar, analysis/uima/**, tools/**, build/**"/>
-      <fileset dir="${basedir}/solr" includes="**/lib/*.jar" excludes="core/lib/*servlet-api*.jar, contrib/analysis-extras/**, test-framework/**, build/**, dist/**, package/**" />
+      <fileset dir="${basedir}/solr" includes="**/lib/*.jar" excludes="core/lib/*servlet-api*.jar, contrib/analysis-extras/**, test-framework/lib/junit*, test-framework/lib/ant*, test-framework/lib/randomizedtesting*, build/**, dist/**, package/**" />
       <map from="${basedir}/" to=""/>
     </pathconvert>
     <xslt in="${ant.file}" out=".classpath" style="dev-tools/eclipse/dot.classpath.xsl" force="true">

lucene/common-build.xml

@@ -354,7 +354,7 @@
   <target name="resolve" depends="ivy-availability-check,ivy-configure">
     <!-- todo, make this a property or something. 
          only special cases need bundles -->
-    <ivy:retrieve type="jar,bundle" log="download-only" 
+    <ivy:retrieve type="jar,bundle,tests" log="download-only" 
                   conf="${ivy.default.configuration}" sync="${ivy.sync}"/>
   </target>
 

lucene/tools/custom-tasks.xml

@@ -43,7 +43,7 @@
 
     <!-- Typical version patterns. -->
     <replaceregex pattern="\.rc[0-9]+" replace="" flags="gi" />
-    <replaceregex pattern="\-(r)?([0-9\-\_\.])+(b(eta)?([0-9\-\.])*)?$" replace="" flags="gi" />
+    <replaceregex pattern="\-(r)?([0-9\-\_\.])+((b(eta)?)|((a(lpha)?))([0-9\-\.])*)?(\-tests)?$" replace="" flags="gi" />
 
     <!-- git hashcode pattern: its always 40 chars right? -->
     <replaceregex pattern="\-[a-z0-9]{40,40}$" replace="" flags="gi" />

lucene/tools/junit4/tests.policy

@@ -59,6 +59,10 @@ grant {
   permission javax.management.MBeanPermission "*", "*";
   permission javax.management.MBeanServerPermission "*";
   permission javax.management.MBeanTrustPermission "*";
+  permission javax.security.auth.AuthPermission "*";
+  permission javax.security.auth.PrivateCredentialPermission "org.apache.hadoop.security.Credentials * \"*\"", "read";
+  permission java.security.SecurityPermission "putProviderProperty.SaslPlainServer";
+  permission java.security.SecurityPermission "insertProvider.SaslPlainServer";
 
   // TIKA uses BouncyCastle and that registers new provider for PDF parsing + MSOffice parsing. Maybe report as bug!
   permission java.security.SecurityPermission "putProviderProperty.BC";

solr/CHANGES.txt

@@ -118,6 +118,9 @@ New Features
 
 * SOLR-4921: Admin UI now supports adding documents to Solr (gsingers, steffkes)
 
+* SOLR-4916: Add support to write and read Solr index files and transaction log
+  files to and from HDFS. (phunt, Mark Miller, Greg Chanan)
+
 Bug Fixes
 ----------------------
 

solr/NOTICE.txt

@@ -52,6 +52,7 @@ including, but not limited to:
  - Apache Jakarta Regexp
  - Apache Commons
  - Apache Xerces
+ - Apache Blur
 
 ICU4J, (under analysis/icu) is licensed under an MIT styles license
 and Copyright (c) 1995-2008 International Business Machines Corporation and others
@@ -72,6 +73,9 @@ http://bitbucket.org/jpbarrette/moman/overview/
 The class org.apache.lucene.util.WeakIdentityMap was derived from
 the Apache CXF project and is Apache License 2.0.
 
+The HdfsDirectory and BlockDirectory were derived from
+the Apache Blur incubating project and are Apache License 2.0.
+
 The Google Code Prettify is Apache License 2.0.
 See http://code.google.com/p/google-code-prettify/
 

solr/build.xml

@@ -40,7 +40,7 @@
   <!-- ========================================================================= -->
 
   <target name="example" description="Creates a runnable example configuration."
-          depends="dist-contrib,dist-war">
+          depends="dist-contrib,dist-war,setup-alt-examples">
     <copy file="${dist}/${fullnamever}.war"
           tofile="${example}/webapps/${ant.project.name}.war"/>
     <jar destfile="${example}/exampledocs/post.jar"
@@ -56,6 +56,15 @@
     <echo>See ${example}/README.txt for how to run the Solr example configuration.</echo>
   </target>
 
+  <target name="setup-alt-examples">
+    <copy todir="${example}/hdfs" overwrite="true">
+      <fileset dir="${example}/solr"/>
+    </copy>
+    <copy todir="${example}/hdfs/collection1/conf" overwrite="true">
+      <fileset dir="${example}/alt-configs/hdfs"/>
+    </copy>
+  </target>
+
   <target name="run-example" depends="example"
           description="Run Solr interactively, via Jetty.  -Dexample.debug=true to enable JVM debugger">
     <property name="example.solr.home" location="example/solr"/>

solr/common-build.xml

@@ -96,6 +96,12 @@
 
   <path id="solr.test.base.classpath">
     <pathelement path="${common-solr.dir}/build/solr-test-framework/classes/java"/>
+    <fileset dir="${common-solr.dir}/test-framework/lib">
+      <include name="*.jar"/>
+      <exclude name="junit-*.jar" />
+      <exclude name="randomizedtesting-runner-*.jar" />
+      <exclude name="ant*.jar" />
+    </fileset>
   	<pathelement path="${build.dir}/test-files"/>
   	<path refid="test.base.classpath"/>
   </path>

solr/core/ivy.xml

@@ -16,6 +16,9 @@
    specific language governing permissions and limitations
    under the License.    
 -->
+<!DOCTYPE ivy-module [
+  <!ENTITY hadoop.version "2.0.5-alpha">
+]>
 <ivy-module version="2.0">
     <info organisation="org.apache.solr" module="core"/>
 
@@ -32,6 +35,14 @@
       <dependency org="javax.servlet" name="javax.servlet-api" rev="3.0.1" transitive="false"/>
       <dependency org="org.restlet.jee" name="org.restlet" rev="2.1.1" transitive="false"/>
       <dependency org="org.restlet.jee" name="org.restlet.ext.servlet" rev="2.1.1" transitive="false"/>
-      <exclude org="*" ext="*" matcher="regexp" type="${ivy.exclude.types}"/>
+
+      <dependency org="org.apache.hadoop" name="hadoop-common" rev="&hadoop.version;" transitive="false"/>
+      <dependency org="org.apache.hadoop" name="hadoop-hdfs" rev="&hadoop.version;" transitive="false"/>
+      <dependency org="org.apache.hadoop" name="hadoop-annotations" rev="&hadoop.version;" transitive="false"/>
+      <dependency org="org.apache.hadoop" name="hadoop-auth" rev="&hadoop.version;" transitive="false"/>
+      <dependency org="commons-configuration" name="commons-configuration" rev="1.6" transitive="false"/>
+      <dependency org="com.google.protobuf" name="protobuf-java" rev="2.4.0a" transitive="false"/>
+      <dependency org="com.googlecode.concurrentlinkedhashmap" name="concurrentlinkedhashmap-lru" rev="1.2" transitive="false"/>
+      <exclude org="*" ext="*" matcher="regexp" type="${ivy.exclude.types}"/> 
     </dependencies>
 </ivy-module>

solr/core/src/java/org/apache/solr/SolrLogFormatter.java

@@ -17,21 +17,25 @@
  * limitations under the License.
  */
 
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.logging.ConsoleHandler;
+import java.util.logging.Formatter;
+import java.util.logging.Handler;
+import java.util.logging.Level;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+
 import org.apache.solr.cloud.ZkController;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.ZkStateReader;
 import org.apache.solr.core.SolrCore;
 import org.apache.solr.request.SolrQueryRequest;
 import org.apache.solr.request.SolrRequestInfo;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.WeakHashMap;
-import java.util.logging.*;
-
 public class SolrLogFormatter extends Formatter {
 
   /** Add this interface to a thread group and the string returned by
@@ -259,7 +263,7 @@ public String _format(LogRecord record) {
 
   private Map<String,Object> getReplicaProps(ZkController zkController, SolrCore core) {
     final String collection = core.getCoreDescriptor().getCloudDescriptor().getCollectionName();
-    Replica replica = zkController.getClusterState().getReplica(collection, zkController.getCoreNodeName(core.getCoreDescriptor()));
+    Replica replica = zkController.getClusterState().getReplica(collection, core.getCoreDescriptor().getCloudDescriptor().getCoreNodeName());
     if(replica!=null) {
       return replica.getProperties();
     }

solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java

@@ -85,6 +85,7 @@ public class JettySolrRunner {
   private String shards;
 
   private String dataDir;
+  private String solrUlogDir;
 
   private volatile boolean startedBefore = false;
 
@@ -359,6 +360,9 @@ public void start(boolean waitForSolr) throws Exception {
     if( dataDir != null) {
       System.setProperty("solr.data.dir", dataDir);
     }
+    if( solrUlogDir != null) {
+      System.setProperty("solr.ulog.dir", solrUlogDir);
+    }
     if(shards != null) {
       System.setProperty("shard", shards);
     }
@@ -382,6 +386,8 @@ public void start(boolean waitForSolr) throws Exception {
     System.clearProperty("shard");
     System.clearProperty("solr.data.dir");
     System.clearProperty("coreNodeName");
+    System.clearProperty("solr.ulog.dir");
+
   }
 
   public void stop() throws Exception {
@@ -485,6 +491,10 @@ public void setShards(String shardList) {
   public void setDataDir(String dataDir) {
     this.dataDir = dataDir;
   }
+
+  public void setUlogDir(String ulogDir) {
+    this.solrUlogDir = ulogDir;
+  }
 
   public void setCoreNodeName(String coreNodeName) {
     this.coreNodeName = coreNodeName;

solr/core/src/java/org/apache/solr/cloud/AssignShard.java → solr/core/src/java/org/apache/solr/cloud/Assign.java

@@ -23,12 +23,36 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.Replica;
 import org.apache.solr.common.cloud.Slice;
 
-public class AssignShard {
 
+public class Assign {
+  private static Pattern COUNT = Pattern.compile("core_node(\\d+)");
+
+  public static String assignNode(String collection, ClusterState state) {
+    Map<String, Slice> sliceMap = state.getSlicesMap(collection);
+    if (sliceMap == null) {
+      return "core_node1";
+    }
+
+    int max = 0;
+    for (Slice slice : sliceMap.values()) {
+      for (Replica replica : slice.getReplicas()) {
+        Matcher m = COUNT.matcher(replica.getName());
+        if (m.matches()) {
+          max = Math.max(max, Integer.parseInt(m.group(1)));
+        }
+      }
+    }
+
+    return "core_node" + (max + 1);
+  }
+
   /**
    * Assign a new unique id up to slices count - then add replicas evenly.
    * 

solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java

@@ -44,6 +44,10 @@ public String getLastPublished() {
   public boolean isLeader() {
     return isLeader;
   }
+
+  public void setLeader(boolean isLeader) {
+    this.isLeader = isLeader;
+  }
 
   public void setShardId(String shardId) {
     this.shardId = shardId;

solr/core/src/java/org/apache/solr/cloud/ElectionContext.java

@@ -47,9 +47,9 @@ public abstract class ElectionContext {
   String leaderSeqPath;
   private SolrZkClient zkClient;
 
-  public ElectionContext(final String shardZkNodeName,
+  public ElectionContext(final String coreNodeName,
       final String electionPath, final String leaderPath, final ZkNodeProps leaderProps, final SolrZkClient zkClient) {
-    this.id = shardZkNodeName;
+    this.id = coreNodeName;
     this.electionPath = electionPath;
     this.leaderPath = leaderPath;
     this.leaderProps = leaderProps;
@@ -78,8 +78,8 @@ class ShardLeaderElectionContextBase extends ElectionContext {
   protected LeaderElector leaderElector;
 
   public ShardLeaderElectionContextBase(LeaderElector leaderElector, final String shardId,
-      final String collection, final String shardZkNodeName, ZkNodeProps props, ZkStateReader zkStateReader) {
-    super(shardZkNodeName, ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + "/leader_elect/"
+      final String collection, final String coreNodeName, ZkNodeProps props, ZkStateReader zkStateReader) {
+    super(coreNodeName, ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + "/leader_elect/"
         + shardId, ZkStateReader.getShardLeadersPath(collection, shardId),
         props, zkStateReader.getZkClient());
     this.leaderElector = leaderElector;
@@ -95,7 +95,7 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException,
     zkClient.makePath(leaderPath, ZkStateReader.toJSON(leaderProps),
         CreateMode.EPHEMERAL, true);
     assert shardId != null;
-    ZkNodeProps m = ZkNodeProps.fromKeyVals(Overseer.QUEUE_OPERATION, "leader",
+    ZkNodeProps m = ZkNodeProps.fromKeyVals(Overseer.QUEUE_OPERATION, ZkStateReader.LEADER_PROP,
         ZkStateReader.SHARD_ID_PROP, shardId, ZkStateReader.COLLECTION_PROP,
         collection, ZkStateReader.BASE_URL_PROP, leaderProps.getProperties()
             .get(ZkStateReader.BASE_URL_PROP), ZkStateReader.CORE_NAME_PROP,
@@ -119,8 +119,8 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
 
   public ShardLeaderElectionContext(LeaderElector leaderElector, 
       final String shardId, final String collection,
-      final String shardZkNodeName, ZkNodeProps props, ZkController zkController, CoreContainer cc) {
-    super(leaderElector, shardId, collection, shardZkNodeName, props,
+      final String coreNodeName, ZkNodeProps props, ZkController zkController, CoreContainer cc) {
+    super(leaderElector, shardId, collection, coreNodeName, props,
         zkController.getZkStateReader());
     this.zkController = zkController;
     this.cc = cc;
@@ -138,12 +138,12 @@ public void close() {
   @Override
   void runLeaderProcess(boolean weAreReplacement) throws KeeperException,
       InterruptedException, IOException {
-    log.info("Running the leader process.");
+    log.info("Running the leader process for shard " + shardId);
 
     String coreName = leaderProps.getStr(ZkStateReader.CORE_NAME_PROP);
 
     // clear the leader in clusterstate
-    ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, "leader",
+    ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, ZkStateReader.LEADER_PROP,
         ZkStateReader.SHARD_ID_PROP, shardId, ZkStateReader.COLLECTION_PROP,
         collection);
     Overseer.getInQueue(zkClient).offer(ZkStateReader.toJSON(m));
@@ -243,8 +243,8 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException,
       }
 
       log.info("I am the new leader: "
-          + ZkCoreNodeProps.getCoreUrl(leaderProps));
-      core.getCoreDescriptor().getCloudDescriptor().isLeader = true;
+          + ZkCoreNodeProps.getCoreUrl(leaderProps) + " " + shardId);
+      core.getCoreDescriptor().getCloudDescriptor().setLeader(true);
     } finally {
       if (core != null) {
         core.close();
@@ -254,16 +254,17 @@ void runLeaderProcess(boolean weAreReplacement) throws KeeperException,
     try {
       super.runLeaderProcess(weAreReplacement);
     } catch (Throwable t) {
+      SolrException.log(log, "There was a problem trying to register as the leader", t);
+      cancelElection();
       try {
         core = cc.getCore(coreName);
         if (core == null) {
-          cancelElection();
           throw new SolrException(ErrorCode.SERVER_ERROR,
               "Fatal Error, SolrCore not found:" + coreName + " in "
                   + cc.getCoreNames());
         }
 
-        core.getCoreDescriptor().getCloudDescriptor().isLeader = false;
+        core.getCoreDescriptor().getCloudDescriptor().setLeader(false);
 
         // we could not publish ourselves as leader - rejoin election
         rejoinLeaderElection(leaderSeqPath, core);
@@ -332,7 +333,7 @@ private void waitForReplicasToComeUp(boolean weAreReplacement,
           return;
         } else {
           if (cnt % 40 == 0) {
-            log.info("Waiting until we see more replicas up: total="
+            log.info("Waiting until we see more replicas up for shard " + shardId + ": total="
               + slices.getReplicasMap().size() + " found=" + found
               + " timeoutin=" + (timeoutAt - System.currentTimeMillis()));
           }