@Test
  public void testFileDumpCorruptDataFiles() throws Exception {
    dropDB(msClient, dbName3);

    // 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);

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

    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName3)
      .withTable(tblName3)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer)
      .withHiveConf(conf)
      .withTransactionBatchSize(10)
      .connect();
    // we need side file for this test, so we create 2 txn batch and test with only one
    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();

    // intentionally corrupt some files
    Path path = new Path(dbLocation);
    Collection<String> files = FileDump.getAllFilesInPath(path, conf);
    for (String file : files) {
      if (file.contains("bucket_00000")) {
        // empty out the file
        corruptDataFile(file, conf, Integer.MIN_VALUE);
      } else if (file.contains("bucket_00001")) {
        corruptDataFile(file, conf, -1);
      } else if (file.contains("bucket_00002")) {
        corruptDataFile(file, conf, 100);
      } else if (file.contains("bucket_00003")) {
        corruptDataFile(file, conf, 100);
      }
    }

    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("Exception"));
    Assert.assertEquals(true, errDump.contains("3 file(s) are corrupted"));
    Assert.assertEquals(false, errDump.contains("is still open for writes."));

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

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

    errDump = new String(myErr.toByteArray());
    Assert.assertEquals(true, errDump.contains("bucket_00001 recovered successfully!"));
    Assert.assertEquals(true, errDump.contains("No readable footers found. Creating empty orc file."));
    Assert.assertEquals(true, errDump.contains("bucket_00002 recovered successfully!"));
    Assert.assertEquals(true, errDump.contains("bucket_00003 recovered successfully!"));
    Assert.assertEquals(false, errDump.contains("Exception"));
    Assert.assertEquals(false, errDump.contains("is still open for writes."));

    // test after recovery
    origErr = System.err;
    myErr = new ByteArrayOutputStream();

    // replace stdout and run command
    System.setErr(new PrintStream(myErr));
    FileDump.main(new String[]{dbLocation});
    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."));

    // after recovery there shouldn'table be any *_flush_length files
    files = FileDump.getAllFilesInPath(path, conf);
    for (String file : files) {
      Assert.assertEquals(false, file.contains("_flush_length"));
    }

    connection.close();
  }
check that transactions that have not heartbeated and timedout get properly aborted

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