Added test and readme update

README.md

@@ -167,6 +167,7 @@ The table below describes all the available configuration values for Waggle Danc
 | `primary-meta-store.hive-metastore-filter-hook`         | No       | Name of the class which implements the `MetaStoreFilterHook` interface from Hive. This allows a metastore filter hook to be applied to the corresponding Hive metastore calls. Can be configured with the `configuration-properties` specified in the `waggle-dance-server.yml` configuration. They will be added in the HiveConf object that is given to the constructor of the `MetaStoreFilterHook` implementation you provide. |
 | `primary-meta-store.database-name-mapping`              | No       | BiDirectional Map of database names and mapped name, where key=`<database name as known in the primary metastore>` and value=`<name that should be shown to a client>`. See the [Database Name Mapping](#database-name-mapping) section.|
 | `primary-meta-store.glue-config`                       | No       | Can be used instead of `remote-meta-store-uris` to federate to an AWS Glue Catalog ([AWS Glue](https://docs.aws.amazon.com/glue/index.html). See the [Federate to AWS Glue Catalog](#federate-to-aws-glue-catalog) section.|
+| `primary-meta-store.read-only-remote-meta-store-uris`    | No       | Can be used to configure an extra read-only endpoint for the primary Metastore. This is an optimization if your environment runs separate Metastore endpoints and traffic needs to be differted efficiently. Waggle Dance will direct traffic to the read-write or read-only endpoints based on the call being done. For instance `get_table` will be a read-only call but `alter_table` will be forwarded to the read-write Metastore.| 
 | `federated-meta-stores`                               | No       | Possible empty list of read only federated metastores. |
 | `federated-meta-stores[n].remote-meta-store-uris`       | Yes      | Thrift URIs of the federated read-only metastore. |
 | `federated-meta-stores[n].name`                        | Yes      | Name that uniquely identifies this metastore. Used internally. Cannot be empty. |

waggle-dance-core/src/test/java/com/hotels/bdp/waggledance/client/SplitTrafficMetastoreClientFactoryTest.java

@@ -0,0 +1,119 @@
+package com.hotels.bdp.waggledance.client;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class SplitTrafficMetastoreClientFactoryTest {
+
+  private @Mock CloseableThriftHiveMetastoreIface readWrite;
+  private @Mock CloseableThriftHiveMetastoreIface readOnly;
+  private @Mock Table readTable;
+  private @Mock Table writeTable;
+
+  @Test
+  public void new_instance_getTable() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+    when(readOnly.get_table("a", "b")).thenReturn(readTable);
+
+    Table table = client.get_table("a", "b");
+
+    assertThat(table, is(readTable));
+    verifyNoInteractions(readWrite);
+  }
+
+  @Test
+  public void new_instance_alterTable() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+
+    client.alter_table("a", "b", writeTable);
+
+    verify(readWrite).alter_table("a", "b", writeTable);
+    verifyNoInteractions(readOnly);
+  }
+
+  @Test
+  public void new_instance_close() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+
+    client.close();
+
+    verify(readWrite).close();
+    verify(readOnly).close();
+  }
+
+  @Test
+  public void new_instance_set_ugi() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+
+    List<String> expected = Arrays.asList("result");
+    when(readOnly.set_ugi("a", Arrays.asList("b"))).thenReturn(expected);
+    when(readWrite.set_ugi("a", Arrays.asList("b"))).thenReturn(expected);
+    List<String> result = client.set_ugi("a", Arrays.asList("b"));
+
+    assertThat(result, is(expected));
+    verify(readOnly).set_ugi("a", Arrays.asList("b"));
+    verify(readWrite).set_ugi("a", Arrays.asList("b"));
+  }
+
+  @Test
+  public void new_instance_isOpen_true() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+    when(readOnly.isOpen()).thenReturn(true);
+    when(readWrite.isOpen()).thenReturn(true);
+    boolean isOpen = client.isOpen();
+
+    assertThat(isOpen, is(true));
+  }
+
+  @Test
+  public void new_instance_isOpen_false() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+    when(readWrite.isOpen()).thenReturn(false);
+    boolean isOpen = client.isOpen();
+
+    assertThat(isOpen, is(false));
+    verifyNoInteractions(readOnly);
+  }
+
+  @Test
+  public void new_instance_isOpen_readOnly_false() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+    when(readOnly.isOpen()).thenReturn(false);
+    when(readWrite.isOpen()).thenReturn(true);
+    boolean isOpen = client.isOpen();
+
+    assertThat(isOpen, is(false));
+  }
+
+  @Test
+  public void new_instance_isOpen_readWrite_false() throws Exception {
+    CloseableThriftHiveMetastoreIface client = new SplitTrafficMetastoreClientFactory()
+        .newInstance(readWrite, readOnly);
+    when(readWrite.isOpen()).thenReturn(false);
+    boolean isOpen = client.isOpen();
+
+    assertThat(isOpen, is(false));
+    verifyNoInteractions(readOnly);
+  }
+
+}