hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ser...@apache.org
Subject [05/25] hive git commit: HIVE-10165 Improve hive-hcatalog-streaming extensibility and support updates and deletes (Eliot West via gates)
Date Thu, 02 Jul 2015 01:49:10 GMT
http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/TestMutations.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/TestMutations.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/TestMutations.java
new file mode 100644
index 0000000..703cef6
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/TestMutations.java
@@ -0,0 +1,544 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 org.apache.hive.hcatalog.streaming.mutate;
+
+import static org.apache.hive.hcatalog.streaming.TransactionBatch.TxnState.ABORTED;
+import static org.apache.hive.hcatalog.streaming.TransactionBatch.TxnState.COMMITTED;
+import static org.apache.hive.hcatalog.streaming.mutate.StreamingTestUtils.databaseBuilder;
+import static org.apache.hive.hcatalog.streaming.mutate.StreamingTestUtils.tableBuilder;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.io.RecordIdentifier;
+import org.apache.hive.hcatalog.streaming.TestStreaming;
+import org.apache.hive.hcatalog.streaming.mutate.StreamingAssert.Factory;
+import org.apache.hive.hcatalog.streaming.mutate.StreamingAssert.Record;
+import org.apache.hive.hcatalog.streaming.mutate.StreamingTestUtils.TableBuilder;
+import org.apache.hive.hcatalog.streaming.mutate.client.MutatorClient;
+import org.apache.hive.hcatalog.streaming.mutate.client.MutatorClientBuilder;
+import org.apache.hive.hcatalog.streaming.mutate.client.AcidTable;
+import org.apache.hive.hcatalog.streaming.mutate.client.Transaction;
+import org.apache.hive.hcatalog.streaming.mutate.worker.BucketIdResolver;
+import org.apache.hive.hcatalog.streaming.mutate.worker.MutatorCoordinator;
+import org.apache.hive.hcatalog.streaming.mutate.worker.MutatorCoordinatorBuilder;
+import org.apache.hive.hcatalog.streaming.mutate.worker.MutatorFactory;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * This test is based on {@link TestStreaming} and has a similar core set of tests to ensure that basic transactional
+ * behaviour is as expected in the {@link RecordMutator} line. This is complemented with a set of tests related to the
+ * use of update and delete operations.
+ */
+public class TestMutations {
+
+  private static final List<String> EUROPE_FRANCE = Arrays.asList("Europe", "France");
+  private static final List<String> EUROPE_UK = Arrays.asList("Europe", "UK");
+  private static final List<String> ASIA_INDIA = Arrays.asList("Asia", "India");
+  // id
+  private static final int[] BUCKET_COLUMN_INDEXES = new int[] { 0 };
+  private static final int RECORD_ID_COLUMN = 2;
+
+  @Rule
+  public TemporaryFolder warehouseFolder = new TemporaryFolder();
+
+  private StreamingTestUtils testUtils = new StreamingTestUtils();
+  private HiveConf conf;
+  private IMetaStoreClient metaStoreClient;
+  private String metaStoreUri;
+  private Database database;
+  private TableBuilder partitionedTableBuilder;
+  private TableBuilder unpartitionedTableBuilder;
+  private Factory assertionFactory;
+
+  public TestMutations() throws Exception {
+    conf = testUtils.newHiveConf(metaStoreUri);
+    testUtils.prepareTransactionDatabase(conf);
+    metaStoreClient = testUtils.newMetaStoreClient(conf);
+    assertionFactory = new StreamingAssert.Factory(metaStoreClient, conf);
+  }
+
+  @Before
+  public void setup() throws Exception {
+    database = databaseBuilder(warehouseFolder.getRoot()).name("testing").dropAndCreate(metaStoreClient);
+
+    partitionedTableBuilder = tableBuilder(database)
+        .name("partitioned")
+        .addColumn("id", "int")
+        .addColumn("msg", "string")
+        .partitionKeys("continent", "country");
+
+    unpartitionedTableBuilder = tableBuilder(database)
+        .name("unpartitioned")
+        .addColumn("id", "int")
+        .addColumn("msg", "string");
+  }
+
+  @Test
+  public void testTransactionBatchEmptyCommitPartitioned() throws Exception {
+    Table table = partitionedTableBuilder.addPartition(ASIA_INDIA).create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    transaction.begin();
+
+    transaction.commit();
+    assertThat(transaction.getState(), is(COMMITTED));
+    client.close();
+  }
+
+  @Test
+  public void testTransactionBatchEmptyCommitUnpartitioned() throws Exception {
+    Table table = unpartitionedTableBuilder.create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), false)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    transaction.begin();
+
+    transaction.commit();
+    assertThat(transaction.getState(), is(COMMITTED));
+    client.close();
+  }
+
+  @Test
+  public void testTransactionBatchEmptyAbortPartitioned() throws Exception {
+    Table table = partitionedTableBuilder.addPartition(ASIA_INDIA).create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    transaction.begin();
+
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+    MutatorCoordinator coordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    coordinator.close();
+
+    transaction.abort();
+    assertThat(transaction.getState(), is(ABORTED));
+    client.close();
+  }
+
+  @Test
+  public void testTransactionBatchEmptyAbortUnartitioned() throws Exception {
+    Table table = unpartitionedTableBuilder.create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), false)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    transaction.begin();
+
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+    MutatorCoordinator coordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    coordinator.close();
+
+    transaction.abort();
+    assertThat(transaction.getState(), is(ABORTED));
+    client.close();
+  }
+
+  @Test
+  public void testTransactionBatchCommitPartitioned() throws Exception {
+    Table table = partitionedTableBuilder.addPartition(ASIA_INDIA).create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    transaction.begin();
+
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+    MutatorCoordinator coordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    BucketIdResolver bucketIdAppender = mutatorFactory.newBucketIdResolver(destinations.get(0).getTotalBuckets());
+    MutableRecord record = (MutableRecord) bucketIdAppender.attachBucketIdToRecord(new MutableRecord(1,
+        "Hello streaming"));
+    coordinator.insert(ASIA_INDIA, record);
+    coordinator.close();
+
+    transaction.commit();
+
+    StreamingAssert streamingAssertions = assertionFactory.newStreamingAssert(table, ASIA_INDIA);
+    streamingAssertions.assertMinTransactionId(1L);
+    streamingAssertions.assertMaxTransactionId(1L);
+    streamingAssertions.assertExpectedFileCount(1);
+
+    List<Record> readRecords = streamingAssertions.readRecords();
+    assertThat(readRecords.size(), is(1));
+    assertThat(readRecords.get(0).getRow(), is("{1, Hello streaming}"));
+    assertThat(readRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 0L)));
+
+    assertThat(transaction.getState(), is(COMMITTED));
+    client.close();
+  }
+
+  @Test
+  public void testMulti() throws Exception {
+    Table table = partitionedTableBuilder.addPartition(ASIA_INDIA).create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    transaction.begin();
+
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+    MutatorCoordinator coordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    BucketIdResolver bucketIdResolver = mutatorFactory.newBucketIdResolver(destinations.get(0).getTotalBuckets());
+    MutableRecord asiaIndiaRecord1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(1,
+        "Hello streaming"));
+    MutableRecord europeUkRecord1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(2,
+        "Hello streaming"));
+    MutableRecord europeFranceRecord1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(3,
+        "Hello streaming"));
+    MutableRecord europeFranceRecord2 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(4,
+        "Bonjour streaming"));
+
+    coordinator.insert(ASIA_INDIA, asiaIndiaRecord1);
+    coordinator.insert(EUROPE_UK, europeUkRecord1);
+    coordinator.insert(EUROPE_FRANCE, europeFranceRecord1);
+    coordinator.insert(EUROPE_FRANCE, europeFranceRecord2);
+    coordinator.close();
+
+    transaction.commit();
+
+    // ASIA_INDIA
+    StreamingAssert streamingAssertions = assertionFactory.newStreamingAssert(table, ASIA_INDIA);
+    streamingAssertions.assertMinTransactionId(1L);
+    streamingAssertions.assertMaxTransactionId(1L);
+    streamingAssertions.assertExpectedFileCount(1);
+
+    List<Record> readRecords = streamingAssertions.readRecords();
+    assertThat(readRecords.size(), is(1));
+    assertThat(readRecords.get(0).getRow(), is("{1, Hello streaming}"));
+    assertThat(readRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 0L)));
+
+    // EUROPE_UK
+    streamingAssertions = assertionFactory.newStreamingAssert(table, EUROPE_UK);
+    streamingAssertions.assertMinTransactionId(1L);
+    streamingAssertions.assertMaxTransactionId(1L);
+    streamingAssertions.assertExpectedFileCount(1);
+
+    readRecords = streamingAssertions.readRecords();
+    assertThat(readRecords.size(), is(1));
+    assertThat(readRecords.get(0).getRow(), is("{2, Hello streaming}"));
+    assertThat(readRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 0L)));
+
+    // EUROPE_FRANCE
+    streamingAssertions = assertionFactory.newStreamingAssert(table, EUROPE_FRANCE);
+    streamingAssertions.assertMinTransactionId(1L);
+    streamingAssertions.assertMaxTransactionId(1L);
+    streamingAssertions.assertExpectedFileCount(1);
+
+    readRecords = streamingAssertions.readRecords();
+    assertThat(readRecords.size(), is(2));
+    assertThat(readRecords.get(0).getRow(), is("{3, Hello streaming}"));
+    assertThat(readRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 0L)));
+    assertThat(readRecords.get(1).getRow(), is("{4, Bonjour streaming}"));
+    assertThat(readRecords.get(1).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 1L)));
+
+    client.close();
+  }
+
+  @Test
+  public void testTransactionBatchCommitUnpartitioned() throws Exception {
+    Table table = unpartitionedTableBuilder.create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), false)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    transaction.begin();
+
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+    MutatorCoordinator coordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    BucketIdResolver bucketIdResolver = mutatorFactory.newBucketIdResolver(destinations.get(0).getTotalBuckets());
+    MutableRecord record = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(1,
+        "Hello streaming"));
+
+    coordinator.insert(Collections.<String> emptyList(), record);
+    coordinator.close();
+
+    transaction.commit();
+
+    StreamingAssert streamingAssertions = assertionFactory.newStreamingAssert(table);
+    streamingAssertions.assertMinTransactionId(1L);
+    streamingAssertions.assertMaxTransactionId(1L);
+    streamingAssertions.assertExpectedFileCount(1);
+
+    List<Record> readRecords = streamingAssertions.readRecords();
+    assertThat(readRecords.size(), is(1));
+    assertThat(readRecords.get(0).getRow(), is("{1, Hello streaming}"));
+    assertThat(readRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 0L)));
+
+    assertThat(transaction.getState(), is(COMMITTED));
+    client.close();
+  }
+
+  @Test
+  public void testTransactionBatchAbort() throws Exception {
+    Table table = partitionedTableBuilder.addPartition(ASIA_INDIA).create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction transaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    transaction.begin();
+
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+    MutatorCoordinator coordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    BucketIdResolver bucketIdResolver = mutatorFactory.newBucketIdResolver(destinations.get(0).getTotalBuckets());
+    MutableRecord record1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(1,
+        "Hello streaming"));
+    MutableRecord record2 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(2,
+        "Welcome to streaming"));
+
+    coordinator.insert(ASIA_INDIA, record1);
+    coordinator.insert(ASIA_INDIA, record2);
+    coordinator.close();
+
+    transaction.abort();
+
+    assertThat(transaction.getState(), is(ABORTED));
+
+    client.close();
+
+    StreamingAssert streamingAssertions = assertionFactory.newStreamingAssert(table, ASIA_INDIA);
+    streamingAssertions.assertNothingWritten();
+  }
+
+  @Test
+  public void testUpdatesAndDeletes() throws Exception {
+    // Set up some base data then stream some inserts/updates/deletes to a number of partitions
+    MutatorFactory mutatorFactory = new ReflectiveMutatorFactory(conf, MutableRecord.class, RECORD_ID_COLUMN,
+        BUCKET_COLUMN_INDEXES);
+
+    // INSERT DATA
+    //
+    Table table = partitionedTableBuilder.addPartition(ASIA_INDIA).addPartition(EUROPE_FRANCE).create(metaStoreClient);
+
+    MutatorClient client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction insertTransaction = client.newTransaction();
+
+    List<AcidTable> destinations = client.getTables();
+
+    insertTransaction.begin();
+
+    MutatorCoordinator insertCoordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    BucketIdResolver bucketIdResolver = mutatorFactory.newBucketIdResolver(destinations.get(0).getTotalBuckets());
+    MutableRecord asiaIndiaRecord1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(1,
+        "Namaste streaming 1"));
+    MutableRecord asiaIndiaRecord2 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(2,
+        "Namaste streaming 2"));
+    MutableRecord europeUkRecord1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(3,
+        "Hello streaming 1"));
+    MutableRecord europeUkRecord2 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(4,
+        "Hello streaming 2"));
+    MutableRecord europeFranceRecord1 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(5,
+        "Bonjour streaming 1"));
+    MutableRecord europeFranceRecord2 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(6,
+        "Bonjour streaming 2"));
+
+    insertCoordinator.insert(ASIA_INDIA, asiaIndiaRecord1);
+    insertCoordinator.insert(ASIA_INDIA, asiaIndiaRecord2);
+    insertCoordinator.insert(EUROPE_UK, europeUkRecord1);
+    insertCoordinator.insert(EUROPE_UK, europeUkRecord2);
+    insertCoordinator.insert(EUROPE_FRANCE, europeFranceRecord1);
+    insertCoordinator.insert(EUROPE_FRANCE, europeFranceRecord2);
+    insertCoordinator.close();
+
+    insertTransaction.commit();
+
+    assertThat(insertTransaction.getState(), is(COMMITTED));
+    client.close();
+
+    // MUTATE DATA
+    //
+    client = new MutatorClientBuilder()
+        .addSinkTable(table.getDbName(), table.getTableName(), true)
+        .metaStoreUri(metaStoreUri)
+        .build();
+    client.connect();
+
+    Transaction mutateTransaction = client.newTransaction();
+
+    destinations = client.getTables();
+
+    mutateTransaction.begin();
+
+    MutatorCoordinator mutateCoordinator = new MutatorCoordinatorBuilder()
+        .metaStoreUri(metaStoreUri)
+        .table(destinations.get(0))
+        .mutatorFactory(mutatorFactory)
+        .build();
+
+    bucketIdResolver = mutatorFactory.newBucketIdResolver(destinations.get(0).getTotalBuckets());
+    MutableRecord asiaIndiaRecord3 = (MutableRecord) bucketIdResolver.attachBucketIdToRecord(new MutableRecord(20,
+        "Namaste streaming 3"));
+
+    mutateCoordinator.update(ASIA_INDIA, new MutableRecord(2, "UPDATED: Namaste streaming 2", new RecordIdentifier(1L,
+        0, 1L)));
+    mutateCoordinator.insert(ASIA_INDIA, asiaIndiaRecord3);
+    mutateCoordinator.delete(EUROPE_UK, new MutableRecord(3, "Hello streaming 1", new RecordIdentifier(1L, 0, 0L)));
+    mutateCoordinator.delete(EUROPE_FRANCE,
+        new MutableRecord(5, "Bonjour streaming 1", new RecordIdentifier(1L, 0, 0L)));
+    mutateCoordinator.update(EUROPE_FRANCE, new MutableRecord(6, "UPDATED: Bonjour streaming 2", new RecordIdentifier(
+        1L, 0, 1L)));
+    mutateCoordinator.close();
+
+    mutateTransaction.commit();
+
+    assertThat(mutateTransaction.getState(), is(COMMITTED));
+
+    StreamingAssert indiaAssertions = assertionFactory.newStreamingAssert(table, ASIA_INDIA);
+    indiaAssertions.assertMinTransactionId(1L);
+    indiaAssertions.assertMaxTransactionId(2L);
+    List<Record> indiaRecords = indiaAssertions.readRecords();
+    assertThat(indiaRecords.size(), is(3));
+    assertThat(indiaRecords.get(0).getRow(), is("{1, Namaste streaming 1}"));
+    assertThat(indiaRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 0L)));
+    assertThat(indiaRecords.get(1).getRow(), is("{2, UPDATED: Namaste streaming 2}"));
+    assertThat(indiaRecords.get(1).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 1L)));
+    assertThat(indiaRecords.get(2).getRow(), is("{20, Namaste streaming 3}"));
+    assertThat(indiaRecords.get(2).getRecordIdentifier(), is(new RecordIdentifier(2L, 0, 0L)));
+
+    StreamingAssert ukAssertions = assertionFactory.newStreamingAssert(table, EUROPE_UK);
+    ukAssertions.assertMinTransactionId(1L);
+    ukAssertions.assertMaxTransactionId(2L);
+    List<Record> ukRecords = ukAssertions.readRecords();
+    assertThat(ukRecords.size(), is(1));
+    assertThat(ukRecords.get(0).getRow(), is("{4, Hello streaming 2}"));
+    assertThat(ukRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 1L)));
+
+    StreamingAssert franceAssertions = assertionFactory.newStreamingAssert(table, EUROPE_FRANCE);
+    franceAssertions.assertMinTransactionId(1L);
+    franceAssertions.assertMaxTransactionId(2L);
+    List<Record> franceRecords = franceAssertions.readRecords();
+    assertThat(franceRecords.size(), is(1));
+    assertThat(franceRecords.get(0).getRow(), is("{6, UPDATED: Bonjour streaming 2}"));
+    assertThat(franceRecords.get(0).getRecordIdentifier(), is(new RecordIdentifier(1L, 0, 1L)));
+
+    client.close();
+  }
+  
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestAcidTableSerializer.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestAcidTableSerializer.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestAcidTableSerializer.java
new file mode 100644
index 0000000..706697a
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestAcidTableSerializer.java
@@ -0,0 +1,66 @@
+package org.apache.hive.hcatalog.streaming.mutate.client;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.junit.Assert.assertThat;
+
+import java.io.File;
+
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hive.hcatalog.streaming.mutate.StreamingTestUtils;
+import org.junit.Test;
+
+public class TestAcidTableSerializer {
+
+  @Test
+  public void testSerializeDeserialize() throws Exception {
+    Database database = StreamingTestUtils.databaseBuilder(new File("/tmp")).name("db_1").build();
+    Table table = StreamingTestUtils
+        .tableBuilder(database)
+        .name("table_1")
+        .addColumn("one", "string")
+        .addColumn("two", "integer")
+        .partitionKeys("partition")
+        .addPartition("p1")
+        .buckets(10)
+        .build();
+
+    AcidTable acidTable = new AcidTable("db_1", "table_1", true, TableType.SINK);
+    acidTable.setTable(table);
+    acidTable.setTransactionId(42L);
+
+    String encoded = AcidTableSerializer.encode(acidTable);
+    System.out.println(encoded);
+    AcidTable decoded = AcidTableSerializer.decode(encoded);
+
+    assertThat(decoded.getDatabaseName(), is("db_1"));
+    assertThat(decoded.getTableName(), is("table_1"));
+    assertThat(decoded.createPartitions(), is(true));
+    assertThat(decoded.getOutputFormatName(), is("org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat"));
+    assertThat(decoded.getTotalBuckets(), is(10));
+    assertThat(decoded.getQualifiedName(), is("DB_1.TABLE_1"));
+    assertThat(decoded.getTransactionId(), is(42L));
+    assertThat(decoded.getTableType(), is(TableType.SINK));
+    assertThat(decoded.getTable(), is(table));
+  }
+
+  @Test
+  public void testSerializeDeserializeNoTableNoTransaction() throws Exception {
+    AcidTable acidTable = new AcidTable("db_1", "table_1", true, TableType.SINK);
+
+    String encoded = AcidTableSerializer.encode(acidTable);
+    AcidTable decoded = AcidTableSerializer.decode(encoded);
+
+    assertThat(decoded.getDatabaseName(), is("db_1"));
+    assertThat(decoded.getTableName(), is("table_1"));
+    assertThat(decoded.createPartitions(), is(true));
+    assertThat(decoded.getOutputFormatName(), is(nullValue()));
+    assertThat(decoded.getTotalBuckets(), is(0));
+    assertThat(decoded.getQualifiedName(), is("DB_1.TABLE_1"));
+    assertThat(decoded.getTransactionId(), is(0L));
+    assertThat(decoded.getTableType(), is(TableType.SINK));
+    assertThat(decoded.getTable(), is(nullValue()));
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestMutatorClient.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestMutatorClient.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestMutatorClient.java
new file mode 100644
index 0000000..ca3f7b2
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestMutatorClient.java
@@ -0,0 +1,176 @@
+package org.apache.hive.hcatalog.streaming.mutate.client;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
+import org.apache.hive.hcatalog.streaming.TransactionBatch.TxnState;
+import org.apache.hive.hcatalog.streaming.mutate.client.lock.Lock;
+import org.apache.hive.hcatalog.streaming.mutate.client.lock.LockFailureListener;
+import org.apache.thrift.TException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestMutatorClient {
+
+  private static final long TRANSACTION_ID = 42L;
+  private static final String TABLE_NAME_1 = "TABLE_1";
+  private static final String TABLE_NAME_2 = "TABLE_2";
+  private static final String DB_NAME = "DB_1";
+  private static final String USER = "user";
+  private static final AcidTable TABLE_1 = new AcidTable(DB_NAME, TABLE_NAME_1, true, TableType.SINK);
+  private static final AcidTable TABLE_2 = new AcidTable(DB_NAME, TABLE_NAME_2, true, TableType.SINK);
+
+  @Mock
+  private IMetaStoreClient mockMetaStoreClient;
+  @Mock
+  private Lock mockLock;
+  @Mock
+  private Table mockTable1, mockTable2;
+  @Mock
+  private StorageDescriptor mockSd;
+  @Mock
+  private Map<String, String> mockParameters;
+  @Mock
+  private HiveConf mockConfiguration;
+  @Mock
+  private LockFailureListener mockLockFailureListener;
+
+  private MutatorClient client;
+
+  @Before
+  public void configureMocks() throws Exception {
+    when(mockMetaStoreClient.getTable(DB_NAME, TABLE_NAME_1)).thenReturn(mockTable1);
+    when(mockTable1.getDbName()).thenReturn(DB_NAME);
+    when(mockTable1.getTableName()).thenReturn(TABLE_NAME_1);
+    when(mockTable1.getSd()).thenReturn(mockSd);
+    when(mockTable1.getParameters()).thenReturn(mockParameters);
+    when(mockMetaStoreClient.getTable(DB_NAME, TABLE_NAME_2)).thenReturn(mockTable2);
+    when(mockTable2.getDbName()).thenReturn(DB_NAME);
+    when(mockTable2.getTableName()).thenReturn(TABLE_NAME_2);
+    when(mockTable2.getSd()).thenReturn(mockSd);
+    when(mockTable2.getParameters()).thenReturn(mockParameters);
+    when(mockSd.getNumBuckets()).thenReturn(1, 2);
+    when(mockSd.getOutputFormat()).thenReturn(OrcOutputFormat.class.getName());
+    when(mockParameters.get("transactional")).thenReturn(Boolean.TRUE.toString());
+
+    when(mockMetaStoreClient.openTxn(USER)).thenReturn(TRANSACTION_ID);
+
+    client = new MutatorClient(mockMetaStoreClient, mockConfiguration, mockLockFailureListener, USER,
+        Collections.singletonList(TABLE_1));
+  }
+
+  @Test
+  public void testCheckValidTableConnect() throws Exception {
+    List<AcidTable> inTables = new ArrayList<>();
+    inTables.add(TABLE_1);
+    inTables.add(TABLE_2);
+    client = new MutatorClient(mockMetaStoreClient, mockConfiguration, mockLockFailureListener, USER, inTables);
+
+    client.connect();
+    List<AcidTable> outTables = client.getTables();
+
+    assertThat(client.isConnected(), is(true));
+    assertThat(outTables.size(), is(2));
+    assertThat(outTables.get(0).getDatabaseName(), is(DB_NAME));
+    assertThat(outTables.get(0).getTableName(), is(TABLE_NAME_1));
+    assertThat(outTables.get(0).getTotalBuckets(), is(2));
+    assertThat(outTables.get(0).getOutputFormatName(), is(OrcOutputFormat.class.getName()));
+    assertThat(outTables.get(0).getTransactionId(), is(0L));
+    assertThat(outTables.get(0).getTable(), is(mockTable1));
+    assertThat(outTables.get(1).getDatabaseName(), is(DB_NAME));
+    assertThat(outTables.get(1).getTableName(), is(TABLE_NAME_2));
+    assertThat(outTables.get(1).getTotalBuckets(), is(2));
+    assertThat(outTables.get(1).getOutputFormatName(), is(OrcOutputFormat.class.getName()));
+    assertThat(outTables.get(1).getTransactionId(), is(0L));
+    assertThat(outTables.get(1).getTable(), is(mockTable2));
+  }
+
+  @Test
+  public void testCheckNonTransactionalTableConnect() throws Exception {
+    when(mockParameters.get("transactional")).thenReturn(Boolean.FALSE.toString());
+
+    try {
+      client.connect();
+      fail();
+    } catch (ConnectionException e) {
+    }
+
+    assertThat(client.isConnected(), is(false));
+  }
+
+  @Test
+  public void testCheckUnBucketedTableConnect() throws Exception {
+    when(mockSd.getNumBuckets()).thenReturn(0);
+
+    try {
+      client.connect();
+      fail();
+    } catch (ConnectionException e) {
+    }
+
+    assertThat(client.isConnected(), is(false));
+  }
+
+  @Test
+  public void testMetaStoreFailsOnConnect() throws Exception {
+    when(mockMetaStoreClient.getTable(anyString(), anyString())).thenThrow(new TException());
+
+    try {
+      client.connect();
+      fail();
+    } catch (ConnectionException e) {
+    }
+
+    assertThat(client.isConnected(), is(false));
+  }
+
+  @Test(expected = ConnectionException.class)
+  public void testGetDestinationsFailsIfNotConnected() throws Exception {
+    client.getTables();
+  }
+
+  @Test
+  public void testNewTransaction() throws Exception {
+    List<AcidTable> inTables = new ArrayList<>();
+    inTables.add(TABLE_1);
+    inTables.add(TABLE_2);
+    client = new MutatorClient(mockMetaStoreClient, mockConfiguration, mockLockFailureListener, USER, inTables);
+
+    client.connect();
+    Transaction transaction = client.newTransaction();
+    List<AcidTable> outTables = client.getTables();
+
+    assertThat(client.isConnected(), is(true));
+
+    assertThat(transaction.getTransactionId(), is(TRANSACTION_ID));
+    assertThat(transaction.getState(), is(TxnState.INACTIVE));
+    assertThat(outTables.get(0).getTransactionId(), is(TRANSACTION_ID));
+    assertThat(outTables.get(1).getTransactionId(), is(TRANSACTION_ID));
+  }
+
+  @Test
+  public void testCloseClosesClient() throws Exception {
+    client.close();
+    assertThat(client.isConnected(), is(false));
+    verify(mockMetaStoreClient).close();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestTransaction.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestTransaction.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestTransaction.java
new file mode 100644
index 0000000..179207a
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/TestTransaction.java
@@ -0,0 +1,95 @@
+package org.apache.hive.hcatalog.streaming.mutate.client;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hive.hcatalog.streaming.TransactionBatch;
+import org.apache.hive.hcatalog.streaming.mutate.client.lock.Lock;
+import org.apache.hive.hcatalog.streaming.mutate.client.lock.LockException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestTransaction {
+
+  private static final String USER = "user";
+  private static final long TRANSACTION_ID = 10L;
+
+  @Mock
+  private Lock mockLock;
+  @Mock
+  private IMetaStoreClient mockMetaStoreClient;
+
+  private Transaction transaction;
+
+  @Before
+  public void createTransaction() throws Exception {
+    when(mockLock.getUser()).thenReturn(USER);
+    when(mockMetaStoreClient.openTxn(USER)).thenReturn(TRANSACTION_ID);
+    transaction = new Transaction(mockMetaStoreClient, mockLock);
+  }
+
+  @Test
+  public void testInitialState() {
+    assertThat(transaction.getState(), is(TransactionBatch.TxnState.INACTIVE));
+    assertThat(transaction.getTransactionId(), is(TRANSACTION_ID));
+  }
+
+  @Test
+  public void testBegin() throws Exception {
+    transaction.begin();
+
+    verify(mockLock).acquire(TRANSACTION_ID);
+    assertThat(transaction.getState(), is(TransactionBatch.TxnState.OPEN));
+  }
+
+  @Test
+  public void testBeginLockFails() throws Exception {
+    doThrow(new LockException("")).when(mockLock).acquire(TRANSACTION_ID);
+
+    try {
+      transaction.begin();
+    } catch (TransactionException ignore) {
+    }
+
+    assertThat(transaction.getState(), is(TransactionBatch.TxnState.INACTIVE));
+  }
+
+  @Test
+  public void testCommit() throws Exception {
+    transaction.commit();
+
+    verify(mockLock).release();
+    verify(mockMetaStoreClient).commitTxn(TRANSACTION_ID);
+    assertThat(transaction.getState(), is(TransactionBatch.TxnState.COMMITTED));
+  }
+
+  @Test(expected = TransactionException.class)
+  public void testCommitLockFails() throws Exception {
+    doThrow(new LockException("")).when(mockLock).release();
+    transaction.commit();
+  }
+
+  @Test
+  public void testAbort() throws Exception {
+    transaction.abort();
+
+    verify(mockLock).release();
+    verify(mockMetaStoreClient).rollbackTxn(TRANSACTION_ID);
+    assertThat(transaction.getState(), is(TransactionBatch.TxnState.ABORTED));
+  }
+
+  @Test(expected = TransactionException.class)
+  public void testAbortLockFails() throws Exception {
+    doThrow(new LockException("")).when(mockLock).release();
+    transaction.abort();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestHeartbeatTimerTask.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestHeartbeatTimerTask.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestHeartbeatTimerTask.java
new file mode 100644
index 0000000..8e6d06e
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestHeartbeatTimerTask.java
@@ -0,0 +1,100 @@
+package org.apache.hive.hcatalog.streaming.mutate.client.lock;
+
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
+import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.TxnAbortedException;
+import org.apache.thrift.TException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestHeartbeatTimerTask {
+
+  private static final long TRANSACTION_ID = 10L;
+  private static final long LOCK_ID = 1L;
+  private static final List<Table> TABLES = createTable();
+
+  @Mock
+  private IMetaStoreClient mockMetaStoreClient;
+  @Mock
+  private LockFailureListener mockListener;
+
+  private HeartbeatTimerTask task;
+
+  @Before
+  public void create() throws Exception {
+    task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, TRANSACTION_ID, TABLES, LOCK_ID);
+  }
+
+  @Test
+  public void testRun() throws Exception {
+    task.run();
+
+    verify(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+  }
+
+  @Test
+  public void testRunNullTransactionId() throws Exception {
+    task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, null, TABLES, LOCK_ID);
+
+    task.run();
+
+    verify(mockMetaStoreClient).heartbeat(0, LOCK_ID);
+  }
+
+  @Test
+  public void testRunHeartbeatFailsNoSuchLockException() throws Exception {
+    NoSuchLockException exception = new NoSuchLockException();
+    doThrow(exception).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+
+    task.run();
+
+    verify(mockListener).lockFailed(LOCK_ID, TRANSACTION_ID, Arrays.asList("DB.TABLE"), exception);
+  }
+
+  @Test
+  public void testRunHeartbeatFailsNoSuchTxnException() throws Exception {
+    NoSuchTxnException exception = new NoSuchTxnException();
+    doThrow(exception).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+
+    task.run();
+
+    verify(mockListener).lockFailed(LOCK_ID, TRANSACTION_ID, Arrays.asList("DB.TABLE"), exception);
+  }
+
+  @Test
+  public void testRunHeartbeatFailsTxnAbortedException() throws Exception {
+    TxnAbortedException exception = new TxnAbortedException();
+    doThrow(exception).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+
+    task.run();
+
+    verify(mockListener).lockFailed(LOCK_ID, TRANSACTION_ID, Arrays.asList("DB.TABLE"), exception);
+  }
+
+  @Test
+  public void testRunHeartbeatFailsTException() throws Exception {
+    TException exception = new TException();
+    doThrow(exception).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+
+    task.run();
+  }
+
+  private static List<Table> createTable() {
+    Table table = new Table();
+    table.setDbName("DB");
+    table.setTableName("TABLE");
+    return Arrays.asList(table);
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestLock.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestLock.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestLock.java
new file mode 100644
index 0000000..ef1e80c
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/client/lock/TestLock.java
@@ -0,0 +1,283 @@
+package org.apache.hive.hcatalog.streaming.mutate.client.lock;
+
+import static org.apache.hadoop.hive.metastore.api.LockState.ABORT;
+import static org.apache.hadoop.hive.metastore.api.LockState.ACQUIRED;
+import static org.apache.hadoop.hive.metastore.api.LockState.NOT_ACQUIRED;
+import static org.apache.hadoop.hive.metastore.api.LockState.WAITING;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import java.net.InetAddress;
+import java.util.Collection;
+import java.util.List;
+import java.util.Timer;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.LockComponent;
+import org.apache.hadoop.hive.metastore.api.LockLevel;
+import org.apache.hadoop.hive.metastore.api.LockRequest;
+import org.apache.hadoop.hive.metastore.api.LockResponse;
+import org.apache.hadoop.hive.metastore.api.LockType;
+import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
+import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.TxnAbortedException;
+import org.apache.thrift.TException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import com.google.common.collect.ImmutableList;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestLock {
+
+  private static final Table TABLE_1 = createTable("DB", "ONE");
+  private static final Table TABLE_2 = createTable("DB", "TWO");
+  private static final List<Table> TABLES = ImmutableList.of(TABLE_1, TABLE_2);
+  private static final long LOCK_ID = 42;
+  private static final long TRANSACTION_ID = 109;
+  private static final String USER = "ewest";
+
+  @Mock
+  private IMetaStoreClient mockMetaStoreClient;
+  @Mock
+  private LockFailureListener mockListener;
+  @Mock
+  private LockResponse mockLockResponse;
+  @Mock
+  private HeartbeatFactory mockHeartbeatFactory;
+  @Mock
+  private Timer mockHeartbeat;
+  @Captor
+  private ArgumentCaptor<LockRequest> requestCaptor;
+
+  private Lock lock;
+  private HiveConf configuration = new HiveConf();
+
+  @Before
+  public void injectMocks() throws Exception {
+    when(mockMetaStoreClient.lock(any(LockRequest.class))).thenReturn(mockLockResponse);
+    when(mockLockResponse.getLockid()).thenReturn(LOCK_ID);
+    when(mockLockResponse.getState()).thenReturn(ACQUIRED);
+    when(
+        mockHeartbeatFactory.newInstance(any(IMetaStoreClient.class), any(LockFailureListener.class), any(Long.class),
+            any(Collection.class), anyLong(), anyInt())).thenReturn(mockHeartbeat);
+
+    lock = new Lock(mockMetaStoreClient, mockHeartbeatFactory, configuration, mockListener, USER, TABLES, 3, 0);
+  }
+
+  @Test
+  public void testAcquireReadLockWithNoIssues() throws Exception {
+    lock.acquire();
+    assertEquals(Long.valueOf(LOCK_ID), lock.getLockId());
+    assertNull(lock.getTransactionId());
+  }
+
+  @Test
+  public void testAcquireTxnLockWithNoIssues() throws Exception {
+    lock.acquire(TRANSACTION_ID);
+    assertEquals(Long.valueOf(LOCK_ID), lock.getLockId());
+    assertEquals(Long.valueOf(TRANSACTION_ID), lock.getTransactionId());
+  }
+
+  @Test
+  public void testAcquireReadLockCheckHeartbeatCreated() throws Exception {
+    configuration.set("hive.txn.timeout", "100s");
+    lock.acquire();
+
+    verify(mockHeartbeatFactory).newInstance(eq(mockMetaStoreClient), eq(mockListener), any(Long.class), eq(TABLES),
+        eq(LOCK_ID), eq(75));
+  }
+
+  @Test
+  public void testAcquireTxnLockCheckHeartbeatCreated() throws Exception {
+    configuration.set("hive.txn.timeout", "100s");
+    lock.acquire(TRANSACTION_ID);
+
+    verify(mockHeartbeatFactory).newInstance(eq(mockMetaStoreClient), eq(mockListener), eq(TRANSACTION_ID), eq(TABLES),
+        eq(LOCK_ID), eq(75));
+  }
+
+  @Test
+  public void testAcquireLockCheckUser() throws Exception {
+    lock.acquire();
+    verify(mockMetaStoreClient).lock(requestCaptor.capture());
+    LockRequest actualRequest = requestCaptor.getValue();
+    assertEquals(USER, actualRequest.getUser());
+  }
+
+  @Test
+  public void testAcquireReadLockCheckLocks() throws Exception {
+    lock.acquire();
+    verify(mockMetaStoreClient).lock(requestCaptor.capture());
+
+    LockRequest request = requestCaptor.getValue();
+    assertEquals(0, request.getTxnid());
+    assertEquals(USER, request.getUser());
+    assertEquals(InetAddress.getLocalHost().getHostName(), request.getHostname());
+
+    List<LockComponent> components = request.getComponent();
+
+    assertEquals(2, components.size());
+
+    LockComponent expected1 = new LockComponent(LockType.SHARED_READ, LockLevel.TABLE, "DB");
+    expected1.setTablename("ONE");
+    assertTrue(components.contains(expected1));
+
+    LockComponent expected2 = new LockComponent(LockType.SHARED_READ, LockLevel.TABLE, "DB");
+    expected2.setTablename("TWO");
+    assertTrue(components.contains(expected2));
+  }
+
+  @Test
+  public void testAcquireTxnLockCheckLocks() throws Exception {
+    lock.acquire(TRANSACTION_ID);
+    verify(mockMetaStoreClient).lock(requestCaptor.capture());
+
+    LockRequest request = requestCaptor.getValue();
+    assertEquals(TRANSACTION_ID, request.getTxnid());
+    assertEquals(USER, request.getUser());
+    assertEquals(InetAddress.getLocalHost().getHostName(), request.getHostname());
+
+    List<LockComponent> components = request.getComponent();
+
+    System.out.println(components);
+    assertEquals(2, components.size());
+
+    LockComponent expected1 = new LockComponent(LockType.SHARED_READ, LockLevel.TABLE, "DB");
+    expected1.setTablename("ONE");
+    assertTrue(components.contains(expected1));
+
+    LockComponent expected2 = new LockComponent(LockType.SHARED_READ, LockLevel.TABLE, "DB");
+    expected2.setTablename("TWO");
+    assertTrue(components.contains(expected2));
+  }
+
+  @Test(expected = LockException.class)
+  public void testAcquireLockNotAcquired() throws Exception {
+    when(mockLockResponse.getState()).thenReturn(NOT_ACQUIRED);
+    lock.acquire();
+  }
+
+  @Test(expected = LockException.class)
+  public void testAcquireLockAborted() throws Exception {
+    when(mockLockResponse.getState()).thenReturn(ABORT);
+    lock.acquire();
+  }
+
+  @Test(expected = LockException.class)
+  public void testAcquireLockWithWaitRetriesExceeded() throws Exception {
+    when(mockLockResponse.getState()).thenReturn(WAITING, WAITING, WAITING);
+    lock.acquire();
+  }
+
+  @Test
+  public void testAcquireLockWithWaitRetries() throws Exception {
+    when(mockLockResponse.getState()).thenReturn(WAITING, WAITING, ACQUIRED);
+    lock.acquire();
+    assertEquals(Long.valueOf(LOCK_ID), lock.getLockId());
+  }
+
+  @Test
+  public void testReleaseLock() throws Exception {
+    lock.acquire();
+    lock.release();
+    verify(mockMetaStoreClient).unlock(LOCK_ID);
+  }
+
+  @Test
+  public void testReleaseLockNoLock() throws Exception {
+    lock.release();
+    verifyNoMoreInteractions(mockMetaStoreClient);
+  }
+
+  @Test
+  public void testReleaseLockCancelsHeartbeat() throws Exception {
+    lock.acquire();
+    lock.release();
+    verify(mockHeartbeat).cancel();
+  }
+
+  @Test
+  public void testReadHeartbeat() throws Exception {
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, null, TABLES, LOCK_ID);
+    task.run();
+    verify(mockMetaStoreClient).heartbeat(0, LOCK_ID);
+  }
+
+  @Test
+  public void testTxnHeartbeat() throws Exception {
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, TRANSACTION_ID, TABLES, LOCK_ID);
+    task.run();
+    verify(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+  }
+
+  @Test
+  public void testReadHeartbeatFailsNoSuchLockException() throws Exception {
+    Throwable t = new NoSuchLockException();
+    doThrow(t).when(mockMetaStoreClient).heartbeat(0, LOCK_ID);
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, null, TABLES, LOCK_ID);
+    task.run();
+    verify(mockListener).lockFailed(LOCK_ID, null, Lock.asStrings(TABLES), t);
+  }
+
+  @Test
+  public void testTxnHeartbeatFailsNoSuchLockException() throws Exception {
+    Throwable t = new NoSuchLockException();
+    doThrow(t).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, TRANSACTION_ID, TABLES, LOCK_ID);
+    task.run();
+    verify(mockListener).lockFailed(LOCK_ID, TRANSACTION_ID, Lock.asStrings(TABLES), t);
+  }
+
+  @Test
+  public void testHeartbeatFailsNoSuchTxnException() throws Exception {
+    Throwable t = new NoSuchTxnException();
+    doThrow(t).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, TRANSACTION_ID, TABLES, LOCK_ID);
+    task.run();
+    verify(mockListener).lockFailed(LOCK_ID, TRANSACTION_ID, Lock.asStrings(TABLES), t);
+  }
+
+  @Test
+  public void testHeartbeatFailsTxnAbortedException() throws Exception {
+    Throwable t = new TxnAbortedException();
+    doThrow(t).when(mockMetaStoreClient).heartbeat(TRANSACTION_ID, LOCK_ID);
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, TRANSACTION_ID, TABLES, LOCK_ID);
+    task.run();
+    verify(mockListener).lockFailed(LOCK_ID, TRANSACTION_ID, Lock.asStrings(TABLES), t);
+  }
+
+  @Test
+  public void testHeartbeatContinuesTException() throws Exception {
+    Throwable t = new TException();
+    doThrow(t).when(mockMetaStoreClient).heartbeat(0, LOCK_ID);
+    HeartbeatTimerTask task = new HeartbeatTimerTask(mockMetaStoreClient, mockListener, TRANSACTION_ID, TABLES, LOCK_ID);
+    task.run();
+    verifyZeroInteractions(mockListener);
+  }
+
+  private static Table createTable(String databaseName, String tableName) {
+    Table table = new Table();
+    table.setDbName(databaseName);
+    table.setTableName(tableName);
+    return table;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestBucketIdResolverImpl.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestBucketIdResolverImpl.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestBucketIdResolverImpl.java
new file mode 100644
index 0000000..f81373e
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestBucketIdResolverImpl.java
@@ -0,0 +1,38 @@
+package org.apache.hive.hcatalog.streaming.mutate.worker;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.hadoop.hive.ql.io.RecordIdentifier;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hive.hcatalog.streaming.mutate.MutableRecord;
+import org.junit.Test;
+
+public class TestBucketIdResolverImpl {
+
+  private static final int TOTAL_BUCKETS = 12;
+  private static final int RECORD_ID_COLUMN = 2;
+  // id - TODO: use a non-zero index to check for offset errors.
+  private static final int[] BUCKET_COLUMN_INDEXES = new int[] { 0 };
+
+  private BucketIdResolver capturingBucketIdResolver = new BucketIdResolverImpl(
+      ObjectInspectorFactory.getReflectionObjectInspector(MutableRecord.class,
+          ObjectInspectorFactory.ObjectInspectorOptions.JAVA), RECORD_ID_COLUMN, TOTAL_BUCKETS, BUCKET_COLUMN_INDEXES);
+
+  @Test
+  public void testAttachBucketIdToRecord() {
+    MutableRecord record = new MutableRecord(1, "hello");
+    capturingBucketIdResolver.attachBucketIdToRecord(record);
+    assertThat(record.rowId, is(new RecordIdentifier(-1L, 8, -1L)));
+    assertThat(record.id, is(1));
+    assertThat(record.msg.toString(), is("hello"));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testNoBucketColumns() {
+    new BucketIdResolverImpl(ObjectInspectorFactory.getReflectionObjectInspector(MutableRecord.class,
+        ObjectInspectorFactory.ObjectInspectorOptions.JAVA), RECORD_ID_COLUMN, TOTAL_BUCKETS, new int[0]);
+
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestGroupingValidator.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestGroupingValidator.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestGroupingValidator.java
new file mode 100644
index 0000000..74fa59b
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestGroupingValidator.java
@@ -0,0 +1,70 @@
+package org.apache.hive.hcatalog.streaming.mutate.worker;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.junit.Test;
+
+public class TestGroupingValidator {
+
+  private GroupingValidator validator = new GroupingValidator();
+
+  @Test
+  public void uniqueGroups() {
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+    assertTrue(validator.isInSequence(Arrays.asList("c", "C"), 3));
+    assertTrue(validator.isInSequence(Arrays.asList("b", "B"), 2));
+  }
+
+  @Test
+  public void sameGroup() {
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+  }
+
+  @Test
+  public void revisitedGroup() {
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+    assertTrue(validator.isInSequence(Arrays.asList("c", "C"), 3));
+    assertFalse(validator.isInSequence(Arrays.asList("a", "A"), 1));
+  }
+
+  @Test
+  public void samePartitionDifferentBucket() {
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+    assertTrue(validator.isInSequence(Arrays.asList("c", "C"), 3));
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 2));
+  }
+
+  @Test
+  public void sameBucketDifferentPartition() {
+    assertTrue(validator.isInSequence(Arrays.asList("a", "A"), 1));
+    assertTrue(validator.isInSequence(Arrays.asList("c", "C"), 3));
+    assertTrue(validator.isInSequence(Arrays.asList("b", "B"), 1));
+  }
+
+  @Test
+  public void uniqueGroupsNoPartition() {
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 1));
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 3));
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 2));
+  }
+
+  @Test
+  public void sameGroupNoPartition() {
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 1));
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 1));
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 1));
+  }
+
+  @Test
+  public void revisitedGroupNoPartition() {
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 1));
+    assertTrue(validator.isInSequence(Collections.<String> emptyList(), 3));
+    assertFalse(validator.isInSequence(Collections.<String> emptyList(), 1));
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorCoordinator.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorCoordinator.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorCoordinator.java
new file mode 100644
index 0000000..6e9ffa2
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorCoordinator.java
@@ -0,0 +1,234 @@
+package org.apache.hive.hcatalog.streaming.mutate.worker;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.ql.io.RecordIdentifier;
+import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
+import org.apache.hive.hcatalog.streaming.mutate.client.AcidTable;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestMutatorCoordinator {
+
+  private static final List<String> UNPARTITIONED = Collections.<String> emptyList();
+  private static final List<String> PARTITION_B = Arrays.asList("B");
+  private static final List<String> PARTITION_A = Arrays.asList("A");
+  private static final long TRANSACTION_ID = 2L;
+  private static final int BUCKET_ID = 0;
+  private static final Path PATH_A = new Path("X");
+  private static final Path PATH_B = new Path("B");
+  private static final Object RECORD = "RECORD";
+  private static final RecordIdentifier ROW__ID_B0_R0 = new RecordIdentifier(10L, BUCKET_ID, 0L);
+  private static final RecordIdentifier ROW__ID_B0_R1 = new RecordIdentifier(10L, BUCKET_ID, 1L);
+  private static final RecordIdentifier ROW__ID_B1_R0 = new RecordIdentifier(10L, BUCKET_ID + 1, 0L);
+  private static final RecordIdentifier ROW__ID_INSERT = new RecordIdentifier(-1L, BUCKET_ID, -1L);
+
+  @Mock
+  private IMetaStoreClient mockMetaStoreClient;
+  @Mock
+  private MutatorFactory mockMutatorFactory;
+  @Mock
+  private CreatePartitionHelper mockPartitionHelper;
+  @Mock
+  private GroupingValidator mockGroupingValidator;
+  @Mock
+  private SequenceValidator mockSequenceValidator;
+  @Mock
+  private AcidTable mockAcidTable;
+  @Mock
+  private RecordInspector mockRecordInspector;
+  @Mock
+  private BucketIdResolver mockBucketIdResolver;
+  @Mock
+  private Mutator mockMutator;
+
+  private MutatorCoordinator coordinator;
+
+  private HiveConf configuration = new HiveConf();
+
+  @Before
+  public void createCoordinator() throws Exception {
+    when(mockAcidTable.getOutputFormatName()).thenReturn(OrcOutputFormat.class.getName());
+    when(mockAcidTable.getTotalBuckets()).thenReturn(1);
+    when(mockAcidTable.getTransactionId()).thenReturn(TRANSACTION_ID);
+    when(mockAcidTable.createPartitions()).thenReturn(true);
+    when(mockMutatorFactory.newRecordInspector()).thenReturn(mockRecordInspector);
+    when(mockMutatorFactory.newBucketIdResolver(anyInt())).thenReturn(mockBucketIdResolver);
+    when(mockMutatorFactory.newMutator(any(OrcOutputFormat.class), anyLong(), any(Path.class), anyInt())).thenReturn(
+        mockMutator);
+    when(mockPartitionHelper.getPathForPartition(any(List.class))).thenReturn(PATH_A);
+    when(mockRecordInspector.extractRecordIdentifier(RECORD)).thenReturn(ROW__ID_INSERT);
+    when(mockSequenceValidator.isInSequence(any(RecordIdentifier.class))).thenReturn(true);
+    when(mockGroupingValidator.isInSequence(any(List.class), anyInt())).thenReturn(true);
+
+    coordinator = new MutatorCoordinator(mockMetaStoreClient, configuration, mockMutatorFactory, mockPartitionHelper,
+        mockGroupingValidator, mockSequenceValidator, mockAcidTable, false);
+  }
+
+  @Test
+  public void insert() throws Exception {
+    coordinator.insert(UNPARTITIONED, RECORD);
+
+    verify(mockPartitionHelper).createPartitionIfNotExists(UNPARTITIONED);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutator).insert(RECORD);
+  }
+
+  @Test
+  public void multipleInserts() throws Exception {
+    coordinator.insert(UNPARTITIONED, RECORD);
+    coordinator.insert(UNPARTITIONED, RECORD);
+    coordinator.insert(UNPARTITIONED, RECORD);
+
+    verify(mockPartitionHelper).createPartitionIfNotExists(UNPARTITIONED);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutator, times(3)).insert(RECORD);
+  }
+
+  @Test
+  public void insertPartitionChanges() throws Exception {
+    when(mockPartitionHelper.getPathForPartition(PARTITION_A)).thenReturn(PATH_A);
+    when(mockPartitionHelper.getPathForPartition(PARTITION_B)).thenReturn(PATH_B);
+
+    coordinator.insert(PARTITION_A, RECORD);
+    coordinator.insert(PARTITION_B, RECORD);
+
+    verify(mockPartitionHelper).createPartitionIfNotExists(PARTITION_A);
+    verify(mockPartitionHelper).createPartitionIfNotExists(PARTITION_B);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_B), eq(BUCKET_ID));
+    verify(mockMutator, times(2)).insert(RECORD);
+  }
+
+  @Test
+  public void bucketChanges() throws Exception {
+    when(mockRecordInspector.extractRecordIdentifier(RECORD)).thenReturn(ROW__ID_B0_R0, ROW__ID_B1_R0);
+
+    when(mockBucketIdResolver.computeBucketId(RECORD)).thenReturn(0, 1);
+
+    coordinator.update(UNPARTITIONED, RECORD);
+    coordinator.delete(UNPARTITIONED, RECORD);
+
+    verify(mockPartitionHelper).createPartitionIfNotExists(UNPARTITIONED);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutatorFactory)
+        .newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID + 1));
+    verify(mockMutator).update(RECORD);
+    verify(mockMutator).delete(RECORD);
+  }
+
+  @Test
+  public void partitionThenBucketChanges() throws Exception {
+    when(mockRecordInspector.extractRecordIdentifier(RECORD)).thenReturn(ROW__ID_B0_R0, ROW__ID_B0_R1, ROW__ID_B1_R0,
+        ROW__ID_INSERT);
+
+    when(mockBucketIdResolver.computeBucketId(RECORD)).thenReturn(0, 0, 1, 0);
+
+    when(mockPartitionHelper.getPathForPartition(PARTITION_A)).thenReturn(PATH_A);
+    when(mockPartitionHelper.getPathForPartition(PARTITION_B)).thenReturn(PATH_B);
+
+    coordinator.update(PARTITION_A, RECORD);
+    coordinator.delete(PARTITION_B, RECORD);
+    coordinator.update(PARTITION_B, RECORD);
+    coordinator.insert(PARTITION_B, RECORD);
+
+    verify(mockPartitionHelper).createPartitionIfNotExists(PARTITION_A);
+    verify(mockPartitionHelper).createPartitionIfNotExists(PARTITION_B);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutatorFactory, times(2)).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_B),
+        eq(BUCKET_ID));
+    verify(mockMutatorFactory)
+        .newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_B), eq(BUCKET_ID + 1));
+    verify(mockMutator, times(2)).update(RECORD);
+    verify(mockMutator).delete(RECORD);
+    verify(mockMutator).insert(RECORD);
+    verify(mockSequenceValidator, times(4)).reset();
+  }
+
+  @Test(expected = RecordSequenceException.class)
+  public void outOfSequence() throws Exception {
+    when(mockSequenceValidator.isInSequence(any(RecordIdentifier.class))).thenReturn(false);
+
+    coordinator.update(UNPARTITIONED, RECORD);
+    coordinator.delete(UNPARTITIONED, RECORD);
+
+    verify(mockPartitionHelper).createPartitionIfNotExists(UNPARTITIONED);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutator).update(RECORD);
+    verify(mockMutator).delete(RECORD);
+  }
+  
+  @Test(expected = GroupRevisitedException.class)
+  public void revisitGroup() throws Exception {
+    when(mockGroupingValidator.isInSequence(any(List.class), anyInt())).thenReturn(false);
+    
+    coordinator.update(UNPARTITIONED, RECORD);
+    coordinator.delete(UNPARTITIONED, RECORD);
+    
+    verify(mockPartitionHelper).createPartitionIfNotExists(UNPARTITIONED);
+    verify(mockMutatorFactory).newMutator(any(OrcOutputFormat.class), eq(TRANSACTION_ID), eq(PATH_A), eq(BUCKET_ID));
+    verify(mockMutator).update(RECORD);
+    verify(mockMutator).delete(RECORD);
+  }
+
+  @Test(expected = BucketIdException.class)
+  public void insertWithBadBucket() throws Exception {
+    when(mockRecordInspector.extractRecordIdentifier(RECORD)).thenReturn(ROW__ID_B0_R0);
+
+    when(mockBucketIdResolver.computeBucketId(RECORD)).thenReturn(1);
+
+    coordinator.insert(UNPARTITIONED, RECORD);
+  }
+
+  @Test(expected = BucketIdException.class)
+  public void updateWithBadBucket() throws Exception {
+    when(mockRecordInspector.extractRecordIdentifier(RECORD)).thenReturn(ROW__ID_B0_R0);
+
+    when(mockBucketIdResolver.computeBucketId(RECORD)).thenReturn(1);
+
+    coordinator.update(UNPARTITIONED, RECORD);
+  }
+
+  @Test
+  public void deleteWithBadBucket() throws Exception {
+    when(mockRecordInspector.extractRecordIdentifier(RECORD)).thenReturn(ROW__ID_B0_R0);
+
+    when(mockBucketIdResolver.computeBucketId(RECORD)).thenReturn(1);
+
+    coordinator.delete(UNPARTITIONED, RECORD);
+  }
+
+  @Test
+  public void closeNoRecords() throws Exception {
+    coordinator.close();
+
+    // No mutator created
+    verifyZeroInteractions(mockMutator);
+  }
+
+  @Test
+  public void closeUsedCoordinator() throws Exception {
+    coordinator.insert(UNPARTITIONED, RECORD);
+    coordinator.close();
+
+    verify(mockMutator).close();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorImpl.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorImpl.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorImpl.java
new file mode 100644
index 0000000..b29c763
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestMutatorImpl.java
@@ -0,0 +1,99 @@
+package org.apache.hive.hcatalog.streaming.mutate.worker;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.io.AcidOutputFormat;
+import org.apache.hadoop.hive.ql.io.AcidOutputFormat.Options;
+import org.apache.hadoop.hive.ql.io.RecordUpdater;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestMutatorImpl {
+
+  private static final Object RECORD = new Object();
+  private static final int RECORD_ID_COLUMN = 2;
+  private static final int BUCKET_ID = 0;
+  private static final Path PATH = new Path("X");
+  private static final long TRANSACTION_ID = 1L;
+
+  @Mock
+  private AcidOutputFormat<?, ?> mockOutputFormat;
+  @Mock
+  private ObjectInspector mockObjectInspector;
+  @Mock
+  private RecordUpdater mockRecordUpdater;
+  @Captor
+  private ArgumentCaptor<AcidOutputFormat.Options> captureOptions;
+
+  private final HiveConf configuration = new HiveConf();
+
+  private Mutator mutator;
+
+  @Before
+  public void injectMocks() throws IOException {
+    when(mockOutputFormat.getRecordUpdater(eq(PATH), any(Options.class))).thenReturn(mockRecordUpdater);
+    mutator = new MutatorImpl(configuration, RECORD_ID_COLUMN, mockObjectInspector, mockOutputFormat, TRANSACTION_ID,
+        PATH, BUCKET_ID);
+  }
+
+  @Test
+  public void testCreatesRecordReader() throws IOException {
+    verify(mockOutputFormat).getRecordUpdater(eq(PATH), captureOptions.capture());
+    Options options = captureOptions.getValue();
+    assertThat(options.getBucket(), is(BUCKET_ID));
+    assertThat(options.getConfiguration(), is((Configuration) configuration));
+    assertThat(options.getInspector(), is(mockObjectInspector));
+    assertThat(options.getRecordIdColumn(), is(RECORD_ID_COLUMN));
+    assertThat(options.getMinimumTransactionId(), is(TRANSACTION_ID));
+    assertThat(options.getMaximumTransactionId(), is(TRANSACTION_ID));
+  }
+
+  @Test
+  public void testInsertDelegates() throws IOException {
+    mutator.insert(RECORD);
+    verify(mockRecordUpdater).insert(TRANSACTION_ID, RECORD);
+  }
+
+  @Test
+  public void testUpdateDelegates() throws IOException {
+    mutator.update(RECORD);
+    verify(mockRecordUpdater).update(TRANSACTION_ID, RECORD);
+  }
+
+  @Test
+  public void testDeleteDelegates() throws IOException {
+    mutator.delete(RECORD);
+    verify(mockRecordUpdater).delete(TRANSACTION_ID, RECORD);
+  }
+
+  @Test
+  public void testCloseDelegates() throws IOException {
+    mutator.close();
+    verify(mockRecordUpdater).close(false);
+  }
+
+  @Test
+  public void testFlushDoesNothing() throws IOException {
+    mutator.flush();
+    verify(mockRecordUpdater, never()).flush();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestRecordInspectorImpl.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestRecordInspectorImpl.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestRecordInspectorImpl.java
new file mode 100644
index 0000000..389ad33
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestRecordInspectorImpl.java
@@ -0,0 +1,31 @@
+package org.apache.hive.hcatalog.streaming.mutate.worker;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.hadoop.hive.ql.io.RecordIdentifier;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
+import org.apache.hive.hcatalog.streaming.mutate.MutableRecord;
+import org.junit.Test;
+
+public class TestRecordInspectorImpl {
+
+  private static final int ROW_ID_COLUMN = 2;
+
+  private RecordInspectorImpl inspector = new RecordInspectorImpl(ObjectInspectorFactory.getReflectionObjectInspector(
+      MutableRecord.class, ObjectInspectorFactory.ObjectInspectorOptions.JAVA), ROW_ID_COLUMN);
+
+  @Test
+  public void testExtractRecordIdentifier() {
+    RecordIdentifier recordIdentifier = new RecordIdentifier(10L, 4, 20L);
+    MutableRecord record = new MutableRecord(1, "hello", recordIdentifier);
+    assertThat(inspector.extractRecordIdentifier(record), is(recordIdentifier));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testNotAStructObjectInspector() {
+    new RecordInspectorImpl(PrimitiveObjectInspectorFactory.javaBooleanObjectInspector, 2);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/994d98c0/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestSequenceValidator.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestSequenceValidator.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestSequenceValidator.java
new file mode 100644
index 0000000..33f9606
--- /dev/null
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/mutate/worker/TestSequenceValidator.java
@@ -0,0 +1,91 @@
+package org.apache.hive.hcatalog.streaming.mutate.worker;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.hadoop.hive.ql.io.RecordIdentifier;
+import org.junit.Test;
+
+public class TestSequenceValidator {
+
+  private static final int BUCKET_ID = 1;
+
+  private SequenceValidator validator = new SequenceValidator();
+
+  @Test
+  public void testSingleInSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+  }
+
+  @Test
+  public void testRowIdInSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 1)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 4)), is(true));
+  }
+
+  @Test
+  public void testTxIdInSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(4L, BUCKET_ID, 0)), is(true));
+  }
+
+  @Test
+  public void testMixedInSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 1)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 1)), is(true));
+  }
+
+  @Test
+  public void testNegativeTxId() {
+    assertThat(validator.isInSequence(new RecordIdentifier(-1L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+  }
+
+  @Test
+  public void testNegativeRowId() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, -1)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+  }
+
+  @Test
+  public void testRowIdOutOfSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 4)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 1)), is(false));
+  }
+
+  @Test
+  public void testReset() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 4)), is(true));
+    // New partition for example
+    validator.reset();
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 1)), is(true));
+  }
+
+  @Test
+  public void testTxIdOutOfSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(4L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 0)), is(false));
+  }
+
+  @Test
+  public void testMixedOutOfSequence() {
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 0)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 4)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 0)), is(false));
+    assertThat(validator.isInSequence(new RecordIdentifier(1L, BUCKET_ID, 5)), is(true));
+    assertThat(validator.isInSequence(new RecordIdentifier(0L, BUCKET_ID, 6)), is(false));
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void testNullRecordIdentifier() {
+    validator.isInSequence(null);
+  }
+
+}


Mime
View raw message