@Test
  public void testFileDump() 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);

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

    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName3)
      .withTable(tblName3)
      .withAgentInfo(agentInfo)
      .withHiveConf(conf)
      .withRecordWriter(writer)
      .connect();
    // 2) Insert data into both tables
    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();

    PrintStream origErr = System.err;
    ByteArrayOutputStream myErr = new ByteArrayOutputStream();

    // replace stderr and run command
    System.setErr(new PrintStream(myErr));
    FileDump.main(new String[]{dbLocation});
    System.err.flush();
    System.setErr(origErr);

    String errDump = new String(myErr.toByteArray());
    Assert.assertEquals(false, errDump.contains("file(s) are corrupted"));
    // since this test runs on local file system which does not have an API to tell if files or
    // open or not, we are testing for negative case even though the bucket files are still open
    // for writes (transaction batch not closed yet)
    Assert.assertEquals(false, errDump.contains("is still open for writes."));

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

    HiveStreamingConnection connection2 = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName4)
      .withTable(tblName4)
      .withAgentInfo(agentInfo)
      .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
    // no data for bucket 3 -- expect 0 length bucket file

    connection2.commitTransaction();
    connection2.close();

    origErr = System.err;
    myErr = new ByteArrayOutputStream();

    // replace stderr and run command
    System.setErr(new PrintStream(myErr));
    FileDump.main(new String[]{dbLocation});
    System.out.flush();
    System.err.flush();
    System.setErr(origErr);

    errDump = new String(myErr.toByteArray());
    Assert.assertEquals(false, errDump.contains("Exception"));
    Assert.assertEquals(false, errDump.contains("file(s) are corrupted"));
    Assert.assertEquals(false, errDump.contains("is still open for writes."));
  }
check that transactions that have not heartbeated and timedout get properly aborted

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