add test for batch indexing from hadoop

integration-tests/README.md

@@ -1,6 +1,20 @@
-Integration Testing
+Integration Testing 
 ===================
 
+To run integration tests, you have to specify the druid cluster the
+tests should use.  
+
+Druid comes with the mvn profile integration-tests
+for setting up druid running in docker containers, and using that
+cluster to run the integration tests.
+
+To use a druid cluster that is already running, use the
+mvn profile int-tests-config-file, which uses a configuration file 
+describing the cluster.
+
+Integration Testing Using Docker 
+-------------------
+
 ## Installing Docker
 
 Please refer to instructions at [https://github.com/druid-io/docker-druid/blob/master/docker-install.md](https://github.com/druid-io/docker-druid/blob/master/docker-install.md).
@@ -20,24 +34,22 @@ eval "$(docker-machine env integration)"
 export DOCKER_IP=$(docker-machine ip integration)
 ```
 
-Running Integration tests
-=========================
+## Running tests
 
-Make sure that you have at least 6GB of memory available before you run the tests.
-
-## Starting docker tests
-
-To run all the tests using docker and mvn run the following command -
+To run all the tests using docker and mvn run the following command:
 ```
   mvn verify -P integration-tests
 ```
 
-To run only a single test using mvn run the following command -
+To run only a single test using mvn run the following command:
 ```
   mvn verify -P integration-tests -Dit.test=<test_name>
 ```
 
-## Configure and run integration tests using existing cluster
+Running Tests Using A Configuration File for Any Cluster
+-------------------
+
+Make sure that you have at least 6GB of memory available before you run the tests.
 
 To run tests on any druid cluster that is already running, create a configuration file:
 
@@ -54,31 +66,92 @@ To run tests on any druid cluster that is already running, create a configuratio
        "zookeeper_hosts": "<comma-separated list of zookeeper_ip:zookeeper_port>",
     }
 
-Set the environment variable CONFIG_FILE to the name of the configuration file -
+Set the environment variable CONFIG_FILE to the name of the configuration file:
 ```
 export CONFIG_FILE=<config file name>
 ```
 
-To run all the tests using mvn run the following command -
+To run all the tests using mvn run the following command: 
 ```
   mvn verify -P int-tests-config-file
 ```
 
-To run only a single test using mvn run the following command -
+To run only a single test using mvn run the following command:
 ```
   mvn verify -P int-tests-config-file -Dit.test=<test_name>
 ```
 
+Running a Test That Uses Hadoop
+-------------------
+
+The integration test that indexes from hadoop is not run as part
+of the integration test run discussed above.  This is because druid
+test clusters might not, in general, have access to hadoop.
+That's the case (for now, at least) when using the docker cluster set 
+up by the integration-tests profile, so the hadoop test
+has to be run using a cluster specified in a configuration file.
+
+The data file is 
+integration-tests/src/test/resources/hadoop/batch_hadoop.data.
+Create a directory called batchHadoop1 in the hadoop file system
+(anywhere you want) and put batch_hadoop.data into that directory
+(as its only file).
+
+Add this keyword to the configuration file (see above):
+
+```
+    "hadoopTestDir": "<name_of_dir_containing_batchHadoop1>"
+```
+
+Run the test using mvn:
+
+```
+  mvn verify -P int-tests-config-file -Dit.test=ITHadoopIndexTest
+```
+
+In some test environments, the machine where the tests need to be executed
+cannot access the outside internet, so mvn cannot be run.  In that case,
+do the following instead of running the tests using mvn:
+
+### Compile druid and the integration tests
+
+On a machine that can do mvn builds:
+
+```
+cd druid 
+mvn clean package
+cd integration_tests 
+mvn dependency:copy-dependencies package
+```
+
+### Put the compiled test code into your test cluster
+
+Copy the integration-tests directory to the test cluster.
+
+### Set CLASSPATH
+
+```
+TDIR=<directory containing integration-tests>/target
+VER=<version of druid you built>
+export CLASSPATH=$TDIR/dependency/*:$TDIR/druid-integration-tests-$VER.jar:$TDIR/druid-integration-tests-$VER-tests.jar
+```
+
+### Run the test
+
+```
+java -Duser.timezone=UTC -Dfile.encoding=UTF-8 -Ddruid.test.config.type=configFile -Ddruid.test.config.configFile=<pathname of configuration file> org.testng.TestNG -testrunfactory org.testng.DruidTestRunnerFactory -testclass io.druid.tests.hadoop.ITHadoopIndexTest
+```
+
 Writing a New Test
-===============
+-------------------
 
 ## What should we cover in integration tests
 
 For every end-user functionality provided by druid we should have an integration-test verifying the correctness.
 
 ## Rules to be followed while writing a new integration test
 
-### Every Integration Test must follow these rules
+### Every Integration Test must follow these rules:
 
 1) Name of the test must start with a prefix "IT"
 2) A test should be independent of other tests
@@ -111,5 +184,5 @@ This will tell the test framework that the test class needs to be constructed us
 Refer ITIndexerTest as an example on how to use dependency Injection
 
 TODOS
-=======================
+-----------------------
 1) Remove the patch for TestNG after resolution of Surefire-622

integration-tests/pom.xml

@@ -45,6 +45,11 @@
             <artifactId>druid-s3-extensions</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.druid.extensions</groupId>
+            <artifactId>druid-datasketches</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.druid.extensions</groupId>
             <artifactId>druid-kafka-eight</artifactId>
@@ -190,6 +195,7 @@
                                 -Dfile.encoding=UTF-8
                                 -Dtestrunfactory=org.testng.DruidTestRunnerFactory
                                 -Ddruid.test.config.dockerIp=${env.DOCKER_IP}
+                                -Ddruid.test.config.hadoopDir=${env.HADOOP_DIR}
                                 -Ddruid.zk.service.host=${env.DOCKER_IP}
                             </argLine>
                             <suiteXmlFiles>

integration-tests/src/main/java/io/druid/testing/DockerConfigProvider.java

@@ -31,6 +31,10 @@ public class DockerConfigProvider implements IntegrationTestingConfigProvider
   @NotNull
   private String dockerIp;
 
+  @JsonProperty
+  @NotNull
+  private String hadoopDir;
+
   @Override
   public IntegrationTestingConfig get()
   {
@@ -87,7 +91,10 @@ public String getKafkaHost()
       @Override
       public String getProperty(String prop)
       {
-        throw new UnsupportedOperationException("DockerConfigProvider does not support getProperty()");
+        if (prop.equals("hadoopTestDir")) {
+	   return hadoopDir;
+	}
+        throw new UnsupportedOperationException("DockerConfigProvider does not support property " + prop);
       }
     };
   }

integration-tests/src/main/java/io/druid/testing/clients/OverlordResourceTestClient.java

@@ -194,6 +194,11 @@ public Map<String, String> shutDownTask(String taskID)
   }
 
   public void waitUntilTaskCompletes(final String taskID)
+  {
+      waitUntilTaskCompletes(taskID, 60000, 10);
+  }
+
+  public void waitUntilTaskCompletes(final String taskID, final int millisEach, final int numTimes)
   {
     RetryUtil.retryUntil(
         new Callable<Boolean>()
@@ -209,9 +214,9 @@ public Boolean call() throws Exception
           }
         },
         true,
-        60000,
-        10,
-        "Index Task to complete"
+        millisEach,
+        numTimes,
+        taskID
     );
   }
 

integration-tests/src/test/java/io/druid/tests/hadoop/ITHadoopIndexTest.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.tests.hadoop;
+
+import com.google.common.base.Throwables;
+import com.google.inject.Inject;
+import com.metamx.common.ISE;
+import com.metamx.common.logger.Logger;
+import io.druid.indexing.common.TaskStatus;
+import io.druid.testing.IntegrationTestingConfig;
+import io.druid.testing.guice.DruidTestModuleFactory;
+import io.druid.testing.utils.RetryUtil;
+import io.druid.tests.indexer.AbstractIndexerTest;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Guice;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.Callable;
+
+@Guice(moduleFactory = DruidTestModuleFactory.class)
+public class ITHadoopIndexTest extends AbstractIndexerTest
+{
+  private static final Logger LOG = new Logger(ITHadoopIndexTest.class);
+  private static final String BATCH_TASK = "/hadoop/batch_hadoop_indexer.json";
+  private static final String BATCH_QUERIES_RESOURCE = "/hadoop/batch_hadoop_queries.json";
+  private static final String BATCH_DATASOURCE = "batchHadoop";
+  private boolean dataLoaded = false;
+
+  @Inject
+  private IntegrationTestingConfig config;
+
+  @BeforeClass
+  public void beforeClass()
+  {
+    loadData(config.getProperty ("hadoopTestDir") + "/batchHadoop1");
+    dataLoaded = true;
+  }
+
+  @Test
+  public void testHadoopIndex() throws Exception
+  {
+      queryHelper.testQueriesFromFile(BATCH_QUERIES_RESOURCE, 2);
+  }
+
+  private void loadData(String hadoopDir)
+  {
+    String indexerSpec = "";
+
+    try {
+       LOG.info("indexerFile name: [%s]", BATCH_TASK);
+       indexerSpec = getTaskAsString(BATCH_TASK);
+       indexerSpec = indexerSpec.replaceAll("%%HADOOP_TEST_PATH%%", hadoopDir);
+    } catch (Exception e) {
+      LOG.error ("could not read and modify indexer file: %s", e.getMessage());
+      throw Throwables.propagate(e);
+    }
+
+    try {
+      final String taskID = indexer.submitTask(indexerSpec);
+      LOG.info("TaskID for loading index task %s", taskID);
+      indexer.waitUntilTaskCompletes (taskID, 60000, 20);
+      RetryUtil.retryUntil(
+          new Callable<Boolean>()
+          {
+            @Override
+            public Boolean call() throws Exception
+            {
+              return coordinator.areSegmentsLoaded(BATCH_DATASOURCE);
+
+            }
+          },
+          true,
+          20000,
+          10,
+          "Segment-Load-Task-" + taskID
+      );
+    } catch (Exception e) {
+      LOG.error ("data could not be loaded: %s", e.getMessage());
+      throw Throwables.propagate(e);
+    }
+  }
+
+  @AfterClass
+  public void afterClass()
+  {
+      if (dataLoaded)
+      {
+         try
+         {
+	     unloadAndKillData(BATCH_DATASOURCE);
+         } catch (Exception e) {
+	     LOG.warn ("exception while removing segments: [%s]", e);
+         }
+      }
+  }
+}