diff --git a/kite-data/kite-data-hcatalog/src/test/java/org/kitesdk/data/hcatalog/TestExternalHCatalogDatasetRepository.java b/kite-data/kite-data-hcatalog/src/test/java/org/kitesdk/data/hcatalog/TestExternalHCatalogDatasetRepository.java
index 25ef601..9f3c15e 100644
--- a/kite-data/kite-data-hcatalog/src/test/java/org/kitesdk/data/hcatalog/TestExternalHCatalogDatasetRepository.java
+++ b/kite-data/kite-data-hcatalog/src/test/java/org/kitesdk/data/hcatalog/TestExternalHCatalogDatasetRepository.java
@@ -16,27 +16,50 @@
 
 package org.kitesdk.data.hcatalog;
 
-import org.kitesdk.data.Dataset;
-import org.kitesdk.data.DatasetDescriptor;
-import org.kitesdk.data.DatasetWriter;
-import org.kitesdk.data.MetadataProvider;
-import org.kitesdk.data.PartitionKey;
-import org.kitesdk.data.PartitionStrategy;
-import org.kitesdk.data.filesystem.TestFileSystemDatasetRepository;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.GenericRecordBuilder;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStore;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.MetaStoreUtils;
 import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.shims.ShimLoader;
 import org.apache.thrift.TException;
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
+import org.kitesdk.data.Dataset;
+import org.kitesdk.data.DatasetDescriptor;
+import org.kitesdk.data.DatasetWriter;
+import org.kitesdk.data.MetadataProvider;
+import org.kitesdk.data.PartitionKey;
+import org.kitesdk.data.PartitionStrategy;
+import org.kitesdk.data.filesystem.TestFileSystemDatasetRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.Socket;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 public class TestExternalHCatalogDatasetRepository extends TestFileSystemDatasetRepository {
 
+  private static final Logger logger = LoggerFactory
+            .getLogger(TestExternalHCatalogDatasetRepository.class);
+
   public TestExternalHCatalogDatasetRepository(boolean distributed) {
     super(distributed);
   }
@@ -46,11 +69,40 @@ public class TestExternalHCatalogDatasetRepository extends TestFileSystemDataset
     return new HCatalogExternalMetadataProvider(conf, testDirectory);
   }
 
+  private File hiveTempDir;
+  private static HiveMetaStore metaStore;
+  private static String metaStoreUri;
   private HiveMetaStoreClient client;
 
+  @BeforeClass
+  public static void startHiveServer() throws Exception {
+      File hiveTempDir = Files.createTempDir();
+      String connectionUrl = String.format("jdbc:derby:;databaseName=%s/metastore_db;create=true", hiveTempDir);
+      System.setProperty("javax.jdo.option.ConnectionURL", connectionUrl);
+      metaStore = new HiveMetaStore();
+      // Using a fixed port simplifies things greatly
+      // hiveMetaStorePort = MetaStoreUtils.findFreePort();
+      int hiveMetaStorePort = 9083;
+      logger.info("Starting MetaStore Server on port {}", hiveMetaStorePort);
+      long before = System.currentTimeMillis();
+      MetaStoreUtils.startMetaStore(hiveMetaStorePort, ShimLoader.getHadoopThriftAuthBridge());
+      waitOnPortInUse(hiveMetaStorePort);
+      logger.info("Starting the hive meta store took {} ms", (System.currentTimeMillis() - before));
+
+      metaStoreUri = "thrift://localhost:" + hiveMetaStorePort;
+      logger.info("Hive Metastore URI = " + metaStoreUri);
+
+  }
+
+  @AfterClass
+  public static void stopHiveServer() throws Exception {
+      // how?
+  }
+
   @Before
-  public void connectToHive() throws MetaException {
-    this.client = new HiveMetaStoreClient(new HiveConf(conf, HiveConf.class));
+  public void connectToHive() throws Exception {
+      conf.set(HiveConf.ConfVars.METASTOREURIS.varname, metaStoreUri);
+      client = new HiveMetaStoreClient(new HiveConf(conf, HiveConf.class));
   }
 
   @After
@@ -69,40 +121,151 @@ public class TestExternalHCatalogDatasetRepository extends TestFileSystemDataset
     HiveTestUtils.assertTableIsExternal(client, "default", NAME);
   }
 
-  @SuppressWarnings("deprecation")
-  @Test
-  public void testNewPartitionIsVisibleToHive() throws Exception {
-    final String NAME2 = "test2";
-
-    PartitionStrategy partitionStrategy = new PartitionStrategy.Builder()
-        .hash("username", 2).build();
-
-    DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
-        .schema(testSchema)
-        .partitionStrategy(partitionStrategy)
-        .build();
-
-    Dataset<GenericRecord> dataset = repo.create(NAME2, descriptor);
-
-    HiveTestUtils.assertTableExists(client, "default", NAME2);
-    HiveTestUtils.assertTableIsExternal(client, "default", NAME2);
-    Assert.assertTrue("No partitions yet",
-        client.listPartitionNames("default", NAME2, (short) 10).isEmpty());
-
-    PartitionKey key0 = partitionStrategy.partitionKey(0);
-    DatasetWriter<GenericRecord> writer = dataset.getPartition(key0, true).newWriter();
-    writer.open();
-    try {
-      GenericRecordBuilder recordBuilder = new GenericRecordBuilder(
-          dataset.getDescriptor().getSchema())
-          .set("username", "0").set("email", "0@example.com");
-      writer.write(recordBuilder.build());
-    } finally {
-      writer.close();
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testNewPartitionIsVisibleToHive() throws Exception {
+        final String NAME2 = "test2";
+
+        PartitionStrategy partitionStrategy = new PartitionStrategy.Builder()
+                .hash("username", 2).build();
+
+        DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
+                .schema(testSchema)
+                .partitionStrategy(partitionStrategy)
+                .build();
+
+        Dataset<GenericRecord> dataset = repo.create(NAME2, descriptor);
+
+        HiveTestUtils.assertTableExists(client, "default", NAME2);
+        HiveTestUtils.assertTableIsExternal(client, "default", NAME2);
+        Assert.assertTrue("No partitions yet",
+                client.listPartitionNames("default", NAME2, (short) 10).isEmpty());
+
+        PartitionKey key0 = partitionStrategy.partitionKey(0);
+        DatasetWriter<GenericRecord> writer = dataset.getPartition(key0, true).newWriter();
+        writer.open();
+        try {
+            GenericRecordBuilder recordBuilder = new GenericRecordBuilder(
+                    dataset.getDescriptor().getSchema())
+                    .set("username", "0").set("email", "0@example.com");
+            writer.write(recordBuilder.build());
+        } finally {
+            writer.close();
+        }
+
+        Assert.assertEquals("Should be one partition", 1,
+                client.listPartitionNames("default", NAME2, (short) 10).size());
+
     }
 
-    Assert.assertEquals("Should be one partition", 1,
-        client.listPartitionNames("default", NAME2, (short) 10).size());
+    /**
+     * This test checks for exceptions during concurrent usage of writers created from a shared
+     * repository. The problem it exposes is in the concurrent usage of the (thrift) HiveMetastoreClient
+     * @throws Exception
+     */
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testConcurrentUse() throws Exception {
+        final String NAME2 = "test2";
+        final String NAME3 = "test3";
+
+        PartitionStrategy partitionStrategy = new PartitionStrategy.Builder()
+                .hash("username", 50).build();
+
+        DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
+                .schema(testSchema)
+                .partitionStrategy(partitionStrategy)
+                .build();
+
+        Dataset<GenericRecord> dataset1 = repo.create(NAME2, descriptor);
+        DatasetWriter<GenericRecord> writer1 = dataset1.newWriter();
+        writer1.open();
+
+        Dataset<GenericRecord> dataset2 = repo.create(NAME3, descriptor);
+        DatasetWriter<GenericRecord> writer2 = dataset2.newWriter();
+        writer2.open();
+
+        try {
+            ExecutorService executors = Executors.newFixedThreadPool(5);
+            List<Callable<Void>> tasks = Lists.newArrayList();
+            int recordsPerWriter = 100;
+            for (int i = 0; i < recordsPerWriter; i++) {
+                GenericRecord record = new GenericRecordBuilder(testSchema)
+                        .set("username", String.valueOf(i)).set("email", String.valueOf(i).concat("@example.com"))
+                        .build();
+                tasks.add(new WriteRecordTask(writer1, "writer 1", i, record));
+                tasks.add(new WriteRecordTask(writer2, "writer 2", i, record));
+            }
+
+            for (Future future: executors.invokeAll(tasks)) {
+                future.get(2, TimeUnit.MINUTES); // this is just here so execeptions from our tasks are thrown
+            }
+
+        } finally {
+            writer1.close();
+            writer2.close();
+        }
+    }
+
+    private class WriteRecordTask implements Callable<Void> {
+
+        private final DatasetWriter<GenericRecord> writer;
+        private final String name;
+        private final int num;
+        private final GenericRecord record;
+
+        public WriteRecordTask(DatasetWriter<GenericRecord> writer, String name, int num, GenericRecord record) {
+            this.writer = writer;
+            this.name = name;
+            this.num = num;
+            this.record = record;
+        }
+
+        @Override
+        public Void call() throws Exception {
+            synchronized(writer) {
+                try {
+                    System.out.println(String.format("%s writing record %d", name, num));
+                    writer.write(record);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    throw e;
+                }
+            }
+            return null;
+        }
+    }
+
+    private static void waitOnPortInUse(int port) throws InterruptedException {
+        long waitUntil = System.currentTimeMillis() + 60000L;
+        while (portAvailable(port)) {
+            if (System.currentTimeMillis() > waitUntil) {
+                throw new RuntimeException("Port " + port + " not in use within timeout.");
+            }
+            Thread.sleep(30);
+        }
+    }
+
+    private static boolean portAvailable(int port) {
+        // Disclaimer: based on http://stackoverflow.com/questions/434718/sockets-discover-port-availability-using-java
+        // answer by TwentyMiles
+        Socket s = null;
+        try {
+            s = new Socket("localhost", port);
+            return false;
+            // If the code makes it this far without an exception it means
+            // something is using the port and has responded.
+        } catch (IOException e) {
+            return true;
+        } finally {
+            if ( s != null) {
+                try {
+                    s.close();
+                } catch (IOException e) {
+                    throw new RuntimeException("You should handle this error." , e);
+                }
+            }
+        }
+    }
 
-  }
 }
