@Test
  public void testBucketing() throws Exception {
    String agentInfo = "UT_" + Thread.currentThread().getName();
    dropDB(msClient, dbName3);
    dropDB(msClient, dbName4);

    // 1) Create two bucketed tables
    String dbLocation = dbFolder.newFolder(dbName3).getCanonicalPath() + ".db";
    dbLocation = dbLocation.replaceAll("\\\\", "/"); // for windows paths
    String[] colNames = "key1,key2,data".split(",");
    String[] colTypes = "string,int,string".split(",");
    String[] bucketNames = "key1,key2".split(",");
    int bucketCount = 4;
    createDbAndTable(driver, dbName3, tblName3, null, colNames, colTypes, bucketNames
      , null, dbLocation, bucketCount);

    String dbLocation2 = dbFolder.newFolder(dbName4).getCanonicalPath() + ".db";
    dbLocation2 = dbLocation2.replaceAll("\\\\", "/"); // for windows paths
    String[] colNames2 = "key3,key4,data2".split(",");
    String[] colTypes2 = "string,int,string".split(",");
    String[] bucketNames2 = "key3,key4".split(",");
    createDbAndTable(driver, dbName4, tblName4, null, colNames2, colTypes2, bucketNames2
      , null, dbLocation2, bucketCount);


    // 2) Insert data into both tables
    StrictDelimitedInputWriter writer = StrictDelimitedInputWriter.newBuilder()
      .withFieldDelimiter(',')
      .build();

    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName3)
      .withTable(tblName3)
      .withAgentInfo(agentInfo)
      .withRecordWriter(writer)
      .withHiveConf(conf)
      .connect();

    connection.beginTransaction();
    connection.write("name0,1,Hello streaming".getBytes());
    connection.write("name2,2,Welcome to streaming".getBytes());
    connection.write("name4,2,more Streaming unlimited".getBytes());
    connection.write("name5,2,even more Streaming unlimited".getBytes());
    connection.commitTransaction();
    connection.close();


    StrictDelimitedInputWriter writer2 = StrictDelimitedInputWriter.newBuilder()
      .withFieldDelimiter(',')
      .build();

    HiveStreamingConnection connection2 = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName4)
      .withTable(tblName4)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer2)
      .withHiveConf(conf)
      .connect();


    connection2.beginTransaction();

    connection2.write("name5,2,fact3".getBytes());  // bucket 0
    connection2.write("name8,2,fact3".getBytes());  // bucket 1
    connection2.write("name0,1,fact1".getBytes());  // bucket 2

    connection2.commitTransaction();

    connection2.close();
    // 3 Check data distribution in  buckets

    HashMap<Integer, ArrayList<SampleRec>> actual1 = dumpAllBuckets(dbLocation, tblName3);
    HashMap<Integer, ArrayList<SampleRec>> actual2 = dumpAllBuckets(dbLocation2, tblName4);
    System.err.println("\n  Table 1");
    System.err.println(actual1);
    System.err.println("\n  Table 2");
    System.err.println(actual2);

    // assert bucket listing is as expected
    Assert.assertEquals("number of buckets does not match expectation", actual1.values().size(), 3);
    Assert.assertTrue("bucket 0 shouldn't have been created", actual1.get(0) == null);
    Assert.assertEquals("records in bucket does not match expectation", actual1.get(1).size(), 1);
    Assert.assertEquals("records in bucket does not match expectation", actual1.get(2).size(), 2);
    Assert.assertEquals("records in bucket does not match expectation", actual1.get(3).size(), 1);
  }
check that transactions that have not heartbeated and timedout get properly aborted

@throws Exception
streaming/src/test/org/apache/hive/streaming/TestStreaming.java