Drill RM configuration changes - Read / Modify / Write configuration (#557)

* Drill RM configuration changes - Read / Modify / Write configuration

Initial code for RM related changes.
- This PR introduces a DrillConfigRenderer interface that should be implemented by an class that represents a Drill configuration and requires the ability to serialize its instance to HOCON format.
- Added DrillRMConfig which represents RM configuration in Java. This can be used read RM config into memory, modify RM config and update the configuration.
- Moved the DrillTestFrameworkUnitTests out of resourcemanagement package.
- Added a sample RM config file to test resources.

* Refactor, cleanup and Unit tests

- Refactored code as a part of cleanup.
- Added unit tests for DrillRMConfig.

TODO:
- Add tests to verify serializer to HOCON

* Add unit tests for DrillRMConfig#render

* Changes to get Drillbit hostnames

- Added utility to get the hostnames on which Drillbits are running on the cluster.
- Added toString to CmdConsOut for better debuggability

* Add utility to apply the RM Config to specified drillbit

* Use TEST_ROOT_DIR instead of CWD

- When running tests from testng, CWD points to rootDir/framework.
- Adding a defualt TEST_ROOT_DIR which always points to root directory.

* Cleanup and add unit tests for the changes.
diff --git a/framework/pom.xml b/framework/pom.xml
index fe52d89..f75c52b 100644
--- a/framework/pom.xml
+++ b/framework/pom.xml
@@ -125,6 +125,11 @@
       <artifactId>testng</artifactId>
       <version>${testng.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.typesafe</groupId>
+      <artifactId>config</artifactId>
+      <version>1.3.2</version>
+    </dependency>
   </dependencies>
   <repositories>
     <repository>
diff --git a/framework/src/main/java/org/apache/drill/test/framework/CmdConsOut.java b/framework/src/main/java/org/apache/drill/test/framework/CmdConsOut.java
index 3503542..bbe3def 100644
--- a/framework/src/main/java/org/apache/drill/test/framework/CmdConsOut.java
+++ b/framework/src/main/java/org/apache/drill/test/framework/CmdConsOut.java
@@ -6,7 +6,6 @@
   public String consoleErr;
   public String cmd;
 
-
   @Override
   public String toString() {
     StringBuilder builder = new StringBuilder("\"CmdConsOut\":{\n");
@@ -35,7 +34,6 @@
             .deleteCharAt(builder.length()-1)
             .deleteCharAt(builder.length()-1)
             .append("}\n");
-    
     return builder.toString();
   }
 }
\ No newline at end of file
diff --git a/framework/src/main/java/org/apache/drill/test/framework/DrillConfigRenderer.java b/framework/src/main/java/org/apache/drill/test/framework/DrillConfigRenderer.java
new file mode 100644
index 0000000..089b1b7
--- /dev/null
+++ b/framework/src/main/java/org/apache/drill/test/framework/DrillConfigRenderer.java
@@ -0,0 +1,152 @@
+package org.apache.drill.test.framework;
+
+import com.google.common.base.Preconditions;
+
+import java.util.List;
+import java.util.stream.IntStream;
+
+/**
+ * This interface should be used by any class that represents a Drill configuration.
+ * The interface defines an implementation for serializing the configuration to HOCON.
+ *
+ * {@link com.typesafe.config.Config} does not have a way to serialize to HOCON format,
+ * which is used by Drill. Hence, it is necessary for the config class to implement this.
+ *
+ * See http://github.com/lightbend/config/issues/453 for more information.
+ *
+ * @see DrillRMConfig
+ * @see DrillRMConfig.SelectorConfig
+ * @see DrillRMConfig.QueueConfig
+ * @see DrillRMConfig.AclConfig
+ *
+ */
+public interface DrillConfigRenderer {
+
+    /**
+     * Returns a representation of the instance in HOCON format.
+     * This method should be used when indentation does not matter.
+     *
+     * @return serialized representation of the class in HOCON format.
+     */
+    String render();
+
+    /**
+     * Returns a representation of the instance in HOCON format.
+     * This method should be used when an instance has to be serialized with indentation.
+     * Mostly for better readability.
+     * @param acc indentation in terms of number of spaces.
+     *
+     * @return serialized representation of the class in HOCON format.
+     */
+    String render(final int acc);
+
+    /**
+     * Defines the format to serialize a config key and value to HOCON format.
+     * The method assumes that there are more configurations to follow.
+     * Hence introduces a "comma (,)" at the very end. If it is the last field,
+     * the caller has to ensure to remove it.
+     *
+     * @param acc indentation in terms of number of spaces.
+     * @param configKey configuration key
+     * @param configVal configuration value
+     * @param <K> type of the key
+     * @param <V> type of the value
+     * @return formatted string for a config key-value pair, in HOCON format.
+     */
+    default <K, V> String formatConfig(final int acc,
+                                final K configKey,
+                                final V configVal) {
+        Preconditions.checkNotNull(configKey, "config key cannot be null!");
+        Preconditions.checkNotNull(configVal, "config value cannot be null!");
+
+        String prefixIndent = indent(acc); //indentation prefix
+
+        StringBuilder sb = new StringBuilder(prefixIndent)
+                .append(configKey)
+                .append(":");
+
+        //String should be within "double quotes".
+        if (configVal instanceof String) {
+            sb.append("\"")
+                    .append(configVal)
+                    .append("\"");
+        } else if (configVal instanceof DrillConfigRenderer) {
+            //If DrillConfigRenderer type, call render.
+            sb.append(((DrillConfigRenderer) configVal).render(acc));
+        } else {
+            sb.append(configVal);
+        }
+
+        sb.append(",");
+        return sb.append("\n").toString();
+    }
+
+    /**
+     * Defines the format to serialize a config key and value to HOCON format
+     * where config value is of type {@link List} .
+     * The method assumes that there are more configurations to follow.
+     * Hence introduces a "comma (,)" at the very end. If it is the last field,
+     * the caller has to ensure to remove it.
+     * @param acc
+     * @param configKey
+     * @param configVal
+     * @param <K>
+     * @param <V>
+     * @return formatted string for a config key-value pair, in HOCON format.
+     */
+    default <K, V> String formatConfig(final int acc,
+                                    final K configKey,
+                                    final List<V> configVal) {
+        Preconditions.checkNotNull(configKey, "config key cannot be null!");
+        Preconditions.checkNotNull(configVal, "config value cannot be null!");
+
+        String prefixIndent = indent(acc); //indentation prefix
+
+        StringBuilder sb = new StringBuilder(prefixIndent)
+                .append(configKey)
+                .append(":")
+                .append("["); //begin square brackets for list
+
+        if(configVal.isEmpty()) { //if empty, close and return
+            return sb.append("]").append(",").append("\n").toString();
+        }
+
+        if (configVal.get(0) instanceof String) {
+            configVal.forEach(val -> sb.append("\"")
+                            .append(val)
+                            .append("\"").append(","));
+            sb.deleteCharAt(sb.length()-1).append("]").append(",");
+        } else if (configVal.get(0) instanceof DrillConfigRenderer) {
+            //Handle values of type DrillConfigRenderer separately
+            //Unlike other types, each value will be rendered in a new line.
+            IntStream
+                    .range(0, configVal.size()-1)
+                    .forEach(index ->
+                            sb.append(((DrillConfigRenderer) configVal.get(index)).render(acc+2))
+                                    .append(",")
+                                    .append("\n")
+                                    .append(indent(acc+2)));
+
+            sb.append(((DrillConfigRenderer) configVal.get(configVal.size()-1)).render(acc+2))
+                    .append("\n").append(prefixIndent).append("]").append(",");
+        } else {
+            configVal.forEach(val -> sb.append(val).append(","));
+            sb.deleteCharAt(sb.length()-1).append("]").append(",");
+        }
+        return sb.append("\n").toString();
+    }
+
+    /**
+     * Return a String with specified indentation. This can be used as a prefix.
+     * @param acc indentation in terms of number of spaces or length of the String.
+     * @return a String of specified length, containing spaces.
+     */
+    default String indent(final int acc) {
+        Preconditions.checkArgument(acc >= 0,
+                "Number of acc for indentation cannot be negative!");
+        StringBuilder sb = new StringBuilder();
+        IntStream.range(0, acc).forEach(i -> sb.append(" "));
+        return sb.toString();
+    }
+
+}
\ No newline at end of file
diff --git a/framework/src/main/java/org/apache/drill/test/framework/DrillRMConfig.java b/framework/src/main/java/org/apache/drill/test/framework/DrillRMConfig.java
new file mode 100644
index 0000000..3c97fc8
--- /dev/null
+++ b/framework/src/main/java/org/apache/drill/test/framework/DrillRMConfig.java
@@ -0,0 +1,288 @@
+package org.apache.drill.test.framework;
+
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigRenderOptions;
+import org.apache.log4j.Logger;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+import static org.apache.drill.test.framework.DrillTestDefaults.DRILL_EXEC_RM_CONFIG_KEY;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeName(DRILL_EXEC_RM_CONFIG_KEY)
+@JsonInclude(Include.NON_DEFAULT)
+/**
+ * Represents a Drill RM Resource Pool configuration.
+ */
+public class DrillRMConfig implements DrillConfigRenderer {
+    private static final Logger LOG = Logger.getLogger(DrillRMConfig.class);
+    //Resource Pool Configurations
+    public static final String RESOURCE_POOL_NAME_KEY = "pool_name";
+    public static final String MEMORY_KEY = "memory";
+    public static final String QUEUE_SELECTION_POLICY_KEY = "queue_selection_policy";
+    public static final String SELECTOR_KEY = "selector";
+    public static final String QUEUE_KEY = "queue";
+    public static final String CHILD_POOLS_KEY = "child_pools";
+
+    //Selector Configurations
+    public static final String SELECTOR_TAG_KEY = "tag";
+    public static final String SELECTOR_ACL_KEY = "acl";
+
+    //ACL Configurations
+    public static final String ACL_USERS_KEY = "users";
+    public static final String ACL_GROUPS_KEY = "groups";
+
+    //Queue Configurations
+    public static final String QUEUE_MAX_QUERY_MEMORY_PER_NODE_KEY = "max_query_memory_per_node";
+    public static final String QUEUE_MAX_WAITING_KEY = "max_waiting";
+    public static final String QUEUE_MAX_ADMISSIBLE_KEY = "max_admissible";
+    public static final String QUEUE_WAIT_FOR_PREFERRED_NODES_KEY = "wait_for_preferred_nodes";
+
+    @JsonProperty(RESOURCE_POOL_NAME_KEY)
+    public String poolName;
+
+    @JsonProperty(MEMORY_KEY)
+    public double memory;
+
+    @JsonProperty(QUEUE_SELECTION_POLICY_KEY)
+    public String queueSelectionPolicy;
+
+    @JsonProperty(SELECTOR_KEY)
+    public SelectorConfig selector;
+
+    @JsonProperty(QUEUE_KEY)
+    public QueueConfig queue;
+
+    @JsonProperty(CHILD_POOLS_KEY)
+    public List<DrillRMConfig> childPools;
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonTypeName(SELECTOR_KEY)
+    @JsonInclude(Include.NON_DEFAULT)
+    /**
+     * Represents a Drill RM Selector configuration.
+     */
+    public static class SelectorConfig implements DrillConfigRenderer {
+
+        public String tag;
+
+        public AclConfig acl;
+
+        @Override
+        public String render() {
+            return render(0);
+        }
+
+        @Override
+        public String render(final int acc) {
+            boolean ensureAtleastOneField = false;
+
+            StringBuilder sb = new StringBuilder("{\n");
+            final int nextAcc = acc+2;
+            if (tag != null) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, SELECTOR_TAG_KEY, tag));
+            }
+
+            if (acl != null) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, SELECTOR_ACL_KEY, acl));
+            }
+
+            if(ensureAtleastOneField) {
+                sb.deleteCharAt(sb.length() - 1)
+                        .deleteCharAt(sb.length() - 1)
+                        .append("\n")
+                        .append(indent(acc)); //Close according to previous level
+            }
+            sb.append("}");
+
+            return sb.toString();
+        }
+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonTypeName(SELECTOR_ACL_KEY)
+    @JsonInclude(Include.NON_DEFAULT)
+    /**
+     * Represents a Drill RM ACL configuration.
+     */
+    public static class AclConfig implements DrillConfigRenderer {
+
+        public List<String> users;
+
+        public List<String> groups;
+
+        @Override
+        public String render() {
+            return render(0);
+        }
+
+        @Override
+        public String render(final int acc) {
+            boolean ensureAtleastOneField = false;
+
+            StringBuilder sb = new StringBuilder("{\n");
+            final int nextAcc = acc+2;
+            if (users != null) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, ACL_USERS_KEY, users));
+            }
+
+            if (groups != null) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, ACL_GROUPS_KEY, groups));
+            }
+
+            if(ensureAtleastOneField) {
+                sb.deleteCharAt(sb.length() - 1)
+                        .deleteCharAt(sb.length() - 1)
+                        .append("\n")
+                        .append(indent(acc)); //Close according to previous level
+            }
+            sb.append("}");
+
+            return sb.toString();
+        }
+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    @JsonInclude(Include.NON_DEFAULT)
+    /**
+     * Represents a Drill RM Queue configuration.
+     */
+    public static class QueueConfig implements DrillConfigRenderer {
+
+        @JsonProperty(QUEUE_MAX_QUERY_MEMORY_PER_NODE_KEY)
+        public long maxQueryMemoryPerNodeInMB;
+
+        @JsonProperty(QUEUE_MAX_WAITING_KEY)
+        public int maxWaitingQueries;
+
+        @JsonProperty(QUEUE_MAX_ADMISSIBLE_KEY)
+        public int maxAdmissibleQueries;
+
+        @JsonProperty(QUEUE_WAIT_FOR_PREFERRED_NODES_KEY)
+        public boolean waitForPreferredNodes;
+
+        @Override
+        public String render() {
+            return render(0);
+        }
+
+        @Override
+        public String render(final int acc) {
+            boolean ensureAtleastOneField = false;
+            StringBuilder sb = new StringBuilder("{\n");
+            final int nextAcc = acc+2;
+
+            if (maxQueryMemoryPerNodeInMB > 0) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, QUEUE_MAX_QUERY_MEMORY_PER_NODE_KEY, maxQueryMemoryPerNodeInMB));
+            }
+
+            if (maxWaitingQueries > 0) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, QUEUE_MAX_WAITING_KEY, maxWaitingQueries));
+            }
+
+            if (maxAdmissibleQueries > 0) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, QUEUE_MAX_ADMISSIBLE_KEY, maxAdmissibleQueries));
+            }
+
+            if (waitForPreferredNodes) {
+                ensureAtleastOneField = true;
+                sb.append(formatConfig(nextAcc, QUEUE_WAIT_FOR_PREFERRED_NODES_KEY, true));
+            }
+
+            if(ensureAtleastOneField) {
+                sb.deleteCharAt(sb.length() - 1)
+                        .deleteCharAt(sb.length() - 1)
+                        .append("\n")
+                        .append(indent(acc)); //Close according to previous level
+            }
+            sb.append("}");
+
+            return sb.toString();
+        }
+    }
+
+    @Override
+    public String render() {
+        return render(0);
+    }
+
+    @Override
+    public String render(final int acc) {
+        boolean ensureAtleastOneField = false;
+        StringBuilder sb = new StringBuilder("{\n");
+        final int nextAcc = acc+2;
+        if (poolName != null) {
+            ensureAtleastOneField = true;
+            sb.append(formatConfig(nextAcc, RESOURCE_POOL_NAME_KEY, poolName));
+        }
+
+        if (memory > 0) {
+            ensureAtleastOneField = true;
+            sb.append(formatConfig(nextAcc, MEMORY_KEY, memory));
+        }
+
+        if (queueSelectionPolicy != null) {
+            ensureAtleastOneField = true;
+            sb.append(formatConfig(nextAcc, QUEUE_SELECTION_POLICY_KEY, queueSelectionPolicy));
+        }
+
+        if (selector != null) {
+            ensureAtleastOneField = true;
+            sb.append(formatConfig(nextAcc, SELECTOR_KEY, selector));
+        }
+
+        if (queue != null) {
+            ensureAtleastOneField = true;
+            sb.append(formatConfig(nextAcc, QUEUE_KEY, queue));
+        }
+
+        if (childPools != null) {
+            ensureAtleastOneField = true;
+            sb.append(formatConfig(nextAcc, CHILD_POOLS_KEY, childPools));
+        }
+
+        if(ensureAtleastOneField) {
+            sb.deleteCharAt(sb.length() - 1)
+                    .deleteCharAt(sb.length() - 1)
+                    .append("\n")
+                    .append(indent(acc)); //Close according to previous level
+        }
+        sb.append("}");
+
+        return sb.toString();
+    }
+
+    public static DrillRMConfig load(final String path) throws IOException {
+        final File file = new File(path);
+
+        if(!file.exists()) {
+            throw new IOException("File " + path + " does not exist");
+        }
+
+        final String rmConfigString = ConfigFactory
+                .parseFile(new File(path))
+                .getConfig(DRILL_EXEC_RM_CONFIG_KEY)
+                .root()
+                .render(ConfigRenderOptions.concise());
+
+        return new ObjectMapper()
+                .readerFor(DrillRMConfig.class)
+                .readValue(rmConfigString);
+    }
+}
diff --git a/framework/src/main/java/org/apache/drill/test/framework/DrillTestDefaults.java b/framework/src/main/java/org/apache/drill/test/framework/DrillTestDefaults.java
index 4f2b2f1..9c86008 100644
--- a/framework/src/main/java/org/apache/drill/test/framework/DrillTestDefaults.java
+++ b/framework/src/main/java/org/apache/drill/test/framework/DrillTestDefaults.java
@@ -69,6 +69,9 @@
   // Current working directory.
   public static final String CWD = System.getProperty("user.dir");
 
+  // Drill Test root directory, CWD changes to test-root/framework for testng
+  public static final String TEST_ROOT_DIR = CWD + (CWD.endsWith("/framework") ? "/../" : "");
+
   // Default user connecting to drill as.
   public static String USERNAME = "root";
 
@@ -104,17 +107,22 @@
 
   // Default line break for logging.
   static final String LINE_BREAK = "------------------------------------------------------------------------";
-  
+
+  //Drill RM config key
+  public static final String DRILL_EXEC_RM_CONFIG_KEY = "drill.exec.rm";
+
+  public static final String DRILL_RM_OVERRIDE_CONF_FILENAME = "drill-rm-override.conf";
+
   // Adding classifications for Execution Failures
   public static enum DRILL_EXCEPTION{
-       VALIDATION_ERROR_INVALID_SCHEMA, 
-       VALIDATION_ERROR_OBJECT_NOT_FOUND, 
-       INTERNAL_ERROR, 
+       VALIDATION_ERROR_INVALID_SCHEMA,
+       VALIDATION_ERROR_OBJECT_NOT_FOUND,
+       INTERNAL_ERROR,
        SQL_EXCEPTION_CONNECTION_ERROR,
-       ALREADY_CLOSED_SQL_EXCEPTION, 
+       ALREADY_CLOSED_SQL_EXCEPTION,
        SQL_EXCEPTION_RESOURCE_ERROR,
-       ARRAY_INDEX_OUT_OF_BOUNDS_EXCEPTION, 
-       UNSUPPORTED_OPERATION_EXCEPTION, 
+       ARRAY_INDEX_OUT_OF_BOUNDS_EXCEPTION,
+       UNSUPPORTED_OPERATION_EXCEPTION,
        SYSTEM_ERROR_ILLEGAL_EXCEPTION,
        SYSTEM_ERROR_ILLEGAL_ARGUMENT_EXCEPTION,
        SYSTEM_ERROR_CHANNEL_CLOSED_EXCEPTION,
@@ -155,7 +163,10 @@
    */
   private static Map<String, String> getConfigProperties() {
     final Map<String, String> properties = Maps.newHashMap();
-    final File overrideFile = new File(CWD + "/conf/" + DRILL_TEST_CONFIG);
+
+    //CWD can be {basePath}/drill-test-framework or {basePath}/drill-test-framework/framework
+    final File overrideFile = new File(TEST_ROOT_DIR + "/conf/"  + DRILL_TEST_CONFIG);
+
     final ResourceBundle bundle;
     if (overrideFile.exists() && !overrideFile.isDirectory()) {
       try {
diff --git a/framework/src/main/java/org/apache/drill/test/framework/TestDriver.java b/framework/src/main/java/org/apache/drill/test/framework/TestDriver.java
index b2ad5de..f985e24 100644
--- a/framework/src/main/java/org/apache/drill/test/framework/TestDriver.java
+++ b/framework/src/main/java/org/apache/drill/test/framework/TestDriver.java
@@ -47,8 +47,6 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.DatabaseMetaData;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 public class TestDriver {
   private static final Logger LOG = Logger.getLogger("DrillTestLogger");
@@ -624,17 +622,17 @@
     }
 
     LOG.info("> Uploading storage plugins");
-    String templatePath = DrillTestDefaults.CWD + "/conf/plugin-templates/common/";
+    String templatePath = DrillTestDefaults.TEST_ROOT_DIR + "/conf/plugin-templates/common/";
     Utils.updateDrillStoragePlugins(templatePath);
 
     if (DrillTestDefaults.IS_SECURE_CLUSTER) {
-      templatePath = DrillTestDefaults.CWD + "/conf/plugin-templates/secure/";
+      templatePath = DrillTestDefaults.TEST_ROOT_DIR + "/conf/plugin-templates/secure/";
     } else {
-      templatePath = DrillTestDefaults.CWD + "/conf/plugin-templates/unsecure/";
+      templatePath = DrillTestDefaults.TEST_ROOT_DIR + "/conf/plugin-templates/unsecure/";
     }
     Utils.updateDrillStoragePlugins(templatePath);
 
-    String beforeRunQueryFilename = DrillTestDefaults.CWD + "/" + cmdParam.beforeRunQueryFilename;
+    String beforeRunQueryFilename = DrillTestDefaults.TEST_ROOT_DIR + "/" + cmdParam.beforeRunQueryFilename;
     LOG.info("\n> Executing init queries\n");
     LOG.info(">> Path: " + beforeRunQueryFilename + "\n");
     try {
@@ -681,7 +679,7 @@
   }
   
   private void teardown() {
-	String afterRunQueryFilename = DrillTestDefaults.CWD + "/" + cmdParam.afterRunQueryFilename;
+	String afterRunQueryFilename = DrillTestDefaults.TEST_ROOT_DIR + "/" + cmdParam.afterRunQueryFilename;
   LOG.info("> Executing queries\n");
   LOG.info(">> Path: " + afterRunQueryFilename + "\n");
 	try {
@@ -740,7 +738,7 @@
           @Override
           public void run() {
             try {
-              Path src = new Path(DrillTestDefaults.CWD + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/" + datasource.src);
+              Path src = new Path(DrillTestDefaults.TEST_ROOT_DIR + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/" + datasource.src);
               Path dest = new Path(DrillTestDefaults.DRILL_TESTDATA, datasource.dest);
               dfsCopy(src, dest, DrillTestDefaults.FS_MODE);
             } catch (IOException e) {
@@ -822,7 +820,7 @@
   }
 
   private void runGenerateScript(DataSource datasource) {
-	String command = DrillTestDefaults.CWD + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/" + datasource.src;
+	String command = DrillTestDefaults.TEST_ROOT_DIR + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/" + datasource.src;
 	LOG.info("Running command " + command);
 	CmdConsOut cmdConsOut;
 	try {
@@ -861,7 +859,7 @@
 			"sum(jvm_direct_current) as jvm_direct_current from sys.memory";
 
 	if (memUsageFilename == null) {
-	  memUsageFilename = Utils.generateOutputFileName(DrillTestDefaults.CWD, "/memComsumption", false);
+	  memUsageFilename = Utils.generateOutputFileName(DrillTestDefaults.TEST_ROOT_DIR, "/memComsumption", false);
 	}
     BufferedWriter writer = new BufferedWriter(new FileWriter(new File(memUsageFilename), true));
     ResultSet resultSet = null;
@@ -922,7 +920,7 @@
         if (!drillReportDir.mkdir()) {
           LOG.debug("Cannot create directory " + DrillTestDefaults.DRILL_REPORTS_DIR
                   + ".  Using current working directory for drill output");
-          DrillTestDefaults.DRILL_REPORTS_DIR = DrillTestDefaults.CWD;
+          DrillTestDefaults.DRILL_REPORTS_DIR = DrillTestDefaults.TEST_ROOT_DIR;
         }
       }
 
@@ -972,7 +970,7 @@
   
   private int restartDrill() {
     int exitCode = 0;
-    String command = DrillTestDefaults.CWD + "/" + DrillTestDefaults.RESTART_DRILL_SCRIPT;
+    String command = DrillTestDefaults.TEST_ROOT_DIR + "/" + DrillTestDefaults.RESTART_DRILL_SCRIPT;
     File commandFile = new File(command);
     if (commandFile.exists() && commandFile.canExecute()) {
       LOG.info("\n> Executing Post Build Script");
diff --git a/framework/src/main/java/org/apache/drill/test/framework/Utils.java b/framework/src/main/java/org/apache/drill/test/framework/Utils.java
index aae1a11..934a4e2 100755
--- a/framework/src/main/java/org/apache/drill/test/framework/Utils.java
+++ b/framework/src/main/java/org/apache/drill/test/framework/Utils.java
@@ -73,6 +73,8 @@
 import javax.net.ssl.SSLSession;
 import javax.net.ssl.TrustManagerFactory;
 
+import static org.apache.drill.test.framework.DrillTestDefaults.*;
+
 /**
  * Collection of utilities supporting the drill test framework.
  *
@@ -224,7 +226,7 @@
       List<File> testDefinitionList = new ArrayList<>();
       if (!absoluteTestDirExpressionFile.exists()) {
         //try regex then exit if failure
-        File drillTestDataDir = new File(DrillTestDefaults.CWD + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR);
+        File drillTestDataDir = new File(DrillTestDefaults.TEST_ROOT_DIR + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR);
         testDefinitionList.addAll(getTestDefinitionList(drillTestDataDir,absoluteTestDirExpression));
         if(testDefinitionList.isEmpty()){
           LOG.info("No regex Found for "+relTestDirExpression);
@@ -491,7 +493,7 @@
     if (filename.startsWith("/")) {
       return filename;
     }
-    return DrillTestDefaults.CWD + "/" + dataDir + "/" + filename;
+    return DrillTestDefaults.TEST_ROOT_DIR + "/" + dataDir + "/" + filename;
   }
 
   /**
@@ -675,7 +677,7 @@
 
   public static void startMinio() {
     LOG.info("> Starting Apache Minio server\n");
-    String cmd = DrillTestDefaults.CWD + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/Datasources/s3minio/minio/run_mn.sh";
+    String cmd = DrillTestDefaults.TEST_ROOT_DIR + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/Datasources/s3minio/minio/run_mn.sh";
     try {
       Runtime.getRuntime().exec(cmd);
     } catch (Throwable e) {
@@ -685,14 +687,16 @@
 
   public static void stopMinio() {
     LOG.info("> Stopping Apache Minio server\n");
-    String cmd = DrillTestDefaults.CWD + "/" + DrillTestDefaults.DRILL_TESTDATA_DIR + "/Datasources/s3minio/minio/stop_mn.sh";
+    String cmd = DrillTestDefaults.TEST_ROOT_DIR + "/" +
+            DrillTestDefaults.DRILL_TESTDATA_DIR + "/Datasources/s3minio/minio/stop_mn.sh";
     try {
       Runtime.getRuntime().exec(cmd);
     } catch (Throwable e) {
       LOG.warn("Fail to run command " + cmd, e);
     }
     LOG.info("> Disabling s3minio storage plugin for Minio\n");
-    String templatePath = DrillTestDefaults.CWD + "/conf/plugin-templates/common/s3minio-storage-plugin.template";
+    String templatePath = DrillTestDefaults.TEST_ROOT_DIR +
+            "/conf/plugin-templates/common/s3minio-storage-plugin.template";
 
     boolean isSuccess = Utils.disableStoragePlugin(templatePath, "s3minio");
     if(!isSuccess){
@@ -891,6 +895,75 @@
 	return numberOfDrillbits;
   }
 
+  /**
+   * Get hostnames of all Drillbits in the cluster.
+   *
+   * @param connection connection instance to the drill cluster
+   * @return a list of hostnames of all Drillbits that form part of the cluster.
+   * @throws SQLException
+   */
+  public static List<String> getDrillbitHosts(Connection connection) throws SQLException {
+    final String columnName = "hostname";
+    final String query = String.format("select %s from sys.drillbits", columnName);
+    PreparedStatement ps = connection.prepareStatement(query);
+    List<String> result = new ArrayList<>();
+    try(ResultSet resultSet = ps.executeQuery()) {
+      while(resultSet.next()) {
+        result.add(resultSet.getString(columnName));
+      }
+    }
+    return result;
+  }
+
+  /**
+   * Apply RM config represented by DrillRMConfig to a specified Drillbit.
+   *
+   * As a part of this method
+   * - Write the config to a temporary file (remove if file exists previously.
+   * - Copy the file to specified Drillbit node.
+   *
+   * @param config
+   * @param drillbitHost
+   * @throws IOException
+   */
+  public static synchronized void applyRMConfigToDrillbit(final DrillRMConfig config,
+                                             final String drillbitHost) throws IOException {
+    final String drillRMConfFilePath = DrillTestDefaults.TEST_ROOT_DIR + "/conf/" + DRILL_RM_OVERRIDE_CONF_FILENAME;
+
+    File drillRMConfFile = new File(drillRMConfFilePath);
+
+    CmdConsOut out;
+    if(drillRMConfFile.exists()) {
+      LOG.warn(drillRMConfFilePath + " exists! Removing the file");
+      if ((out = Utils.execCmd("rm -rf " + drillRMConfFilePath)).exitCode != 0) {
+        LOG.error("Could not remove config file " +
+                drillRMConfFilePath + "\n\n" +
+                out);
+        throw new IOException(out.consoleErr);
+      }
+    }
+
+    try (BufferedWriter writer = new BufferedWriter(new FileWriter(drillRMConfFilePath))) {
+      writer.write(DRILL_EXEC_RM_CONFIG_KEY + ":" + config.render());
+    }
+
+    final String scpCommand = new StringBuilder("scp ")
+            .append(drillRMConfFilePath)
+            .append(" ")
+            .append(USERNAME)
+            .append("@").append(drillbitHost)
+            .append(":").append(DRILL_HOME)
+            .append("/conf/")
+            .append(DRILL_RM_OVERRIDE_CONF_FILENAME)
+            .toString();
+
+    LOG.info("Copying config " + scpCommand);
+    if ((out = Utils.execCmd(scpCommand)).exitCode != 0) {
+      LOG.error("Copying config to drillbit failed!\n\n" + out);
+      throw new IOException(out.consoleErr);
+    }
+  }
+
   public static boolean sanityTest(Connection connection) {
     String query = "SELECT count(*) FROM cp.`employee.json`";
     int rowCount = 0;
diff --git a/framework/src/test/java/org/apache/drill/test/framework/DrillTestFrameworkUnitTests.java b/framework/src/test/java/org/apache/drill/test/framework/DrillTestFrameworkUnitTests.java
new file mode 100644
index 0000000..afe2ee7
--- /dev/null
+++ b/framework/src/test/java/org/apache/drill/test/framework/DrillTestFrameworkUnitTests.java
@@ -0,0 +1,162 @@
+package org.apache.drill.test.framework;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.drill.test.framework.common.DrillJavaTestBase;
+import org.apache.log4j.Logger;
+import org.testng.Assert;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Properties;
+
+import static org.apache.drill.test.framework.DrillTestDefaults.DRILL_EXEC_RM_CONFIG_KEY;
+import static org.apache.drill.test.framework.common.DrillTestNGDefaults.UNIT_GROUP;
+
+@Test(groups = UNIT_GROUP)
+public class DrillTestFrameworkUnitTests extends DrillJavaTestBase {
+    private static final Logger LOG = Logger.getLogger(DrillTestFrameworkUnitTests.class);
+    private static final String SAMPLE_RM_CONFIG_NAME =
+            DrillTestDefaults.CWD + "/src/test/resources/sample-drill-rm-override.conf";
+
+    @BeforeTest(alwaysRun = true)
+    public void runBeforeTest() {
+        LOG.debug("Running before test");
+    }
+
+
+    /**
+     * A unit test to validate that {@link Utils#getQueryProfile(String)}
+     * can be used to obtain query profile. This test also validates that
+     * the returned query profile can be deserialized to {@link DrillQueryProfile}.
+     *
+     * @param method
+     */
+    @Test(groups = UNIT_GROUP)
+    public void testGetQueryProfile(Method method) {
+        final Properties props = Utils.createConnectionProperties();
+        final ConnectionPool pool = new ConnectionPool(props);
+        final String sqlStatement = "select name, val, status from sys.options where name like \'%runtime%\'";
+
+        try (Connection connection = pool.getOrCreateConnection()) {
+            Statement statement = connection.createStatement();
+            ResultSet resultSet = statement.executeQuery(sqlStatement);
+
+            final String queryId = Utils.getQueryID(resultSet);
+            DrillQueryProfile profile = Utils.getQueryProfile(queryId);
+            Assert.assertEquals(profile.queryId, queryId);
+        } catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail("Test " + method.getName() + " failed due to " + e.getMessage());
+        }
+    }
+
+    /**
+     * A unit test to validate that {@link Utils#getQueryProfile(String)}
+     * fails with the right error if a profile for a specified queryId does not exist.
+     */
+    @Test(groups = UNIT_GROUP)
+    public void testQueryProfileDoesNotExist() {
+        final String queryId = "invalidQueryId";
+
+        try {
+            Utils.getQueryProfile(queryId);
+            Assert.fail("Querying invalid query profile did not throw exception!");
+        } catch (IOException e) {
+            Assert.assertTrue(e.getMessage().contains("Could not get query profile"),
+                    "Expected error message \"Could not get query profile\" " +
+                            "but obtained - " + e.getMessage());
+        }
+    }
+
+
+    /**
+     * Test reading a sample RM config file in to a Java Bean.
+     */
+    @Test(groups = UNIT_GROUP)
+    public void testReadSampleRMConfigFile() {
+        try {
+            DrillRMConfig drillRMConfig = DrillRMConfig.load(SAMPLE_RM_CONFIG_NAME);
+            Assert.assertEquals(drillRMConfig.poolName, "root",
+                    "Root resource pool name did not match");
+
+            Assert.assertEquals(drillRMConfig.childPools.size(), 2,
+                    "Number of child pools in the config did not match!");
+        } catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+    /**
+     * Negative test to validate the behavior when the config file does not exist.
+     */
+    @Test(groups = UNIT_GROUP, expectedExceptions = IOException.class)
+    public void testLoadConfigWhenFileDoesNotExist() throws IOException {
+        final String invalidPath = "invalid-config-file.conf";
+
+        DrillRMConfig.load(invalidPath);
+    }
+
+    /**
+     * Test reading a sample RM config file in to a Java Bean.
+     */
+    @Test(groups = UNIT_GROUP)
+    public void testConfigFileRenderer() {
+        try {
+            DrillRMConfig drillRMConfig = DrillRMConfig.load(SAMPLE_RM_CONFIG_NAME);
+            Assert.assertEquals(drillRMConfig.poolName, "root",
+                    "Root resource pool name did not match");
+
+            Assert.assertEquals(drillRMConfig.childPools.size(), 2,
+                    "Number of child pools in the config did not match!");
+
+            Config config = ConfigFactory.parseString(drillRMConfig.render());
+            Assert.assertEquals(config.getString(DrillRMConfig.RESOURCE_POOL_NAME_KEY),
+                    drillRMConfig.poolName,
+                    "Pool names did not match!");
+            Assert.assertEquals(config.getString(DrillRMConfig.QUEUE_SELECTION_POLICY_KEY),
+                    drillRMConfig.queueSelectionPolicy,
+                    "Queue selection policy did not match!");
+        } catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+    /**
+     * Test if DrillRMConfig can be serialized to a file.
+     * Read from the file to validate.
+     * @throws IOException
+     */
+    public void testWriteRMConfigToFile() throws IOException {
+        final String fileName = "tempRMConfig.conf";
+        final String filePath = DrillTestDefaults.TEST_ROOT_DIR + "conf/" + fileName;
+        File file = new File(filePath);
+
+        if(file.exists()) {
+            LOG.warn(filePath + " exists! Removing the file");
+            Utils.execCmd("rm -rf " + filePath);
+        }
+
+        try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath))) {
+            DrillRMConfig drillRMConfig = DrillRMConfig.load(SAMPLE_RM_CONFIG_NAME);
+            writer.write(DRILL_EXEC_RM_CONFIG_KEY + ":" + drillRMConfig.render());
+        }
+
+        DrillRMConfig drillRMConfig2 = DrillRMConfig.load(filePath);
+        Assert.assertEquals(drillRMConfig2.poolName, "root",
+                "Root resource pool name did not match");
+
+        Assert.assertEquals(drillRMConfig2.childPools.size(), 2,
+                "Number of child pools in the config did not match!");
+    }
+}
diff --git a/framework/src/test/java/org/apache/drill/test/framework/common/DrillJavaTestBase.java b/framework/src/test/java/org/apache/drill/test/framework/common/DrillJavaTestBase.java
index 09e3ce0..0fa102c 100644
--- a/framework/src/test/java/org/apache/drill/test/framework/common/DrillJavaTestBase.java
+++ b/framework/src/test/java/org/apache/drill/test/framework/common/DrillJavaTestBase.java
@@ -10,6 +10,8 @@
 import org.testng.annotations.BeforeSuite;
 import org.testng.annotations.BeforeTest;
 
+import java.lang.reflect.Method;
+
 public class DrillJavaTestBase {
     private static final Logger LOG = Logger.getLogger(DrillJavaTestBase.class);
 
@@ -29,13 +31,13 @@
     }
 
     @BeforeMethod(alwaysRun = true, description = "Invoked before every Test Method.")
-    public void baseBeforeMethod() {
-        LOG.debug("Running Base Before Method");
+    public void baseBeforeMethod(Method method) {
+        LOG.info("\n\n---------- Test " + method.getName() + " started ----------\n\n");
     }
 
     @AfterMethod(alwaysRun = true, description = "Invoked after every Test Method")
-    public void baseAfterMethod() {
-        LOG.debug("Running Base After Method");
+    public void baseAfterMethod(Method method) {
+        LOG.info("\n\n---------- Test " + method.getName() + " finished ----------\n\n");
     }
 
     @AfterClass(alwaysRun = true, description = "Invoked after all tests in a Test Class finish.")
diff --git a/framework/src/test/java/org/apache/drill/test/framework/resourcemanagement/DrillTestFrameworkUnitTests.java b/framework/src/test/java/org/apache/drill/test/framework/resourcemanagement/DrillTestFrameworkUnitTests.java
deleted file mode 100644
index da7dd9d..0000000
--- a/framework/src/test/java/org/apache/drill/test/framework/resourcemanagement/DrillTestFrameworkUnitTests.java
+++ /dev/null
@@ -1,78 +0,0 @@
-package org.apache.drill.test.framework.resourcemanagement;
-
-import org.apache.drill.test.framework.ConnectionPool;
-import org.apache.drill.test.framework.DrillQueryProfile;
-import org.apache.drill.test.framework.Utils;
-import org.apache.drill.test.framework.common.DrillJavaTestBase;
-import org.apache.log4j.Logger;
-import org.testng.Assert;
-import org.testng.annotations.BeforeTest;
-import org.testng.annotations.Test;
-
-import java.io.IOException;
-import java.lang.reflect.Method;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.Statement;
-import java.util.Properties;
-
-import static org.apache.drill.test.framework.common.DrillTestNGDefaults.UNIT_GROUP;
-
-@Test(groups = UNIT_GROUP)
-public class DrillTestFrameworkUnitTests extends DrillJavaTestBase {
-    private static final Logger LOG = Logger.getLogger(DrillTestFrameworkUnitTests.class);
-
-    @BeforeTest(alwaysRun = true)
-    public void runBeforeTest() {
-        LOG.debug("Running before test");
-    }
-    
-    /**
-     * A unit test to validate that {@link Utils#getQueryProfile(String)}
-     * can be used to obtain query profile. This test also validates that
-     * the returned query profile can be deserialized to {@link DrillQueryProfile}.
-     *
-     * @param method
-     */
-    @Test(groups = UNIT_GROUP)
-    public void testGetQueryProfile(Method method) {
-        LOG.info("Test " + method.getName() + " started.");
-
-        final Properties props = Utils.createConnectionProperties();
-        final ConnectionPool pool = new ConnectionPool(props);
-        final String sqlStatement = "select name, val, status from sys.options where name like \'%runtime%\'";
-
-        try (Connection connection = pool.getOrCreateConnection()) {
-            Statement statement = connection.createStatement();
-            ResultSet resultSet = statement.executeQuery(sqlStatement);
-
-            final String queryId = Utils.getQueryID(resultSet);
-            DrillQueryProfile profile = Utils.getQueryProfile(queryId);
-            Assert.assertEquals(profile.queryId, queryId);
-        } catch (Exception e) {
-            e.printStackTrace();
-            Assert.fail("Test " + method.getName() + " failed due to " + e.getMessage());
-        }
-    }
-
-    /**
-     * A unit test to validate that {@link Utils#getQueryProfile(String)}
-     * fails with the right error if a profile for a specified queryId does not exist.
-     *
-     * @param method
-     */
-    @Test(groups = UNIT_GROUP)
-    public void testQueryProfileDoesNotExist(Method method) {
-        LOG.info("Test " + method.getName() + " started.");
-        final String queryId = "invalidQueryId";
-
-        try {
-            Utils.getQueryProfile(queryId);
-            Assert.fail("Querying invalid query profile did not throw exception!");
-        } catch (IOException e) {
-            Assert.assertTrue(e.getMessage().contains("Could not get query profile"),
-                    "Expected error message \"Could not get query profile\" " +
-                            "but obtained - " + e.getMessage());
-        }
-    }
-}
diff --git a/framework/src/test/resources/sample-drill-rm-override.conf b/framework/src/test/resources/sample-drill-rm-override.conf
new file mode 100644
index 0000000..e11d9e5
--- /dev/null
+++ b/framework/src/test/resources/sample-drill-rm-override.conf
@@ -0,0 +1,67 @@
+drill.exec.rm:{
+  pool_name:"root",
+  memory:1.00,
+  queue_selection_policy:"bestfit",
+  child_pools:[
+    {
+      pool_name:"Devs Resource Pool",
+      memory:0.75,
+      selector:{
+        acl:{
+          groups:["dev"]
+        }
+      },
+      child_pools:[
+        {
+          pool_name:"Small Query Resource Pool",
+          memory:0.20,
+          selector:{
+            tag:"small"
+          },
+          queue:{
+            max_query_memory_per_node:6881,
+            max_waiting:5,
+            max_wait_timeout:2000
+          }
+        },
+        {
+          pool_name:"Large Query Resource Pool",
+          memory:0.60,
+          selector:{
+            tag:"large"
+          },
+          queue:{
+            max_query_memory_per_node:5505,
+            max_waiting:2
+          }
+        },
+        {
+          pool_name:"Experimental Query Resource Pool",
+          memory:0.20,
+          selector:{
+            tag :"experimental"
+          },
+          queue:{
+            max_query_memory_per_node:4587,
+            max_waiting:4
+          }
+        }
+      ]
+    },
+    {
+      pool_name:"Marketing Resource Pool",
+      memory:0.25,
+      selector:{
+        acl :{
+          groups:["marketing"],
+          users:["bob"]
+        }
+      },
+      queue:{
+        max_query_memory_per_node:4587,
+        max_waiting:10,
+        wait_for_preferred_nodes:false
+      }
+    }
+  ]
+}
\ No newline at end of file
diff --git a/framework/testng-java.xml b/framework/testng-java.xml
index 3876755..0afc56c 100644
--- a/framework/testng-java.xml
+++ b/framework/testng-java.xml
@@ -2,11 +2,11 @@
 <suite name="Drill Java Test Suite">
     <test name="Resource Management Tests">
         <classes>
-            <class name="org.apache.drill.test.framework.resourcemanagement.DrillTestFrameworkUnitTests" />
+            <class name="org.apache.drill.test.framework.DrillTestFrameworkUnitTests" />
         </classes>
         <groups>
             <run>
-                <include name="functional" />
+                <include name="unit" />
                 <exclude name="broken" />
             </run>
         </groups>