@Test
  public void testInterleavedTransactionBatchCommits() throws Exception {
    StrictDelimitedInputWriter writer = StrictDelimitedInputWriter.newBuilder()
      .withFieldDelimiter(',')
      .build();

    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName)
      .withTable(tblName)
      .withStaticPartitionValues(partitionVals)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer)
      .withHiveConf(conf)
      .withTransactionBatchSize(10)
      .connect();

    // Acquire 1st Txn Batch
    connection.beginTransaction();

    // Acquire 2nd Txn Batch
    StrictDelimitedInputWriter writer2 = StrictDelimitedInputWriter.newBuilder()
      .withFieldDelimiter(',')
      .build();
    HiveStreamingConnection connection2 = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName)
      .withTable(tblName)
      .withStaticPartitionValues(partitionVals)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer2)
      .withHiveConf(conf)
      .withTransactionBatchSize(10)
      .connect();
    connection2.beginTransaction();

    // Interleaved writes to both batches
    connection.write("1,Hello streaming".getBytes());
    connection2.write("3,Hello streaming - once again".getBytes());

    checkNothingWritten(partLoc);

    connection2.commitTransaction();

    String validationQuery = "select id, msg from " + dbName + "." + tblName + " order by id, msg";
    checkDataWritten2(partLoc, 11, 20, 1,
      validationQuery, true, "3\tHello streaming - once again");

    connection.commitTransaction();
    /*now both batches have committed (but not closed) so we for each primary file we expect a side
    file to exist and indicate the true length of primary file*/
    FileSystem fs = partLoc.getFileSystem(conf);
    AcidDirectory dir = AcidUtils.getAcidState(fs, partLoc, conf, getTransactionContext(conf), null, false);
    for (AcidUtils.ParsedDelta pd : dir.getCurrentDirectories()) {
      for (FileStatus stat : fs.listStatus(pd.getPath(), AcidUtils.bucketFileFilter)) {
        Path lengthFile = OrcAcidUtils.getSideFile(stat.getPath());
        Assert.assertTrue(lengthFile + " missing", fs.exists(lengthFile));
        long lengthFileSize = fs.getFileStatus(lengthFile).getLen();
        Assert.assertTrue("Expected " + lengthFile + " to be non empty. lengh=" +
          lengthFileSize, lengthFileSize > 0);
        long logicalLength = AcidUtils.getLogicalLength(fs, stat);
        long actualLength = stat.getLen();
        Assert.assertTrue("", logicalLength == actualLength);
      }
    }
    checkDataWritten2(partLoc, 1, 20, 2,
      validationQuery, false, "1\tHello streaming", "3\tHello streaming - once again");

    connection.beginTransaction();
    connection.write("2,Welcome to streaming".getBytes());

    connection2.beginTransaction();
    connection2.write("4,Welcome to streaming - once again".getBytes());
    //here each batch has written data and committed (to bucket0 since table only has 1 bucket)
    //so each of 2 deltas has 1 bucket0 and 1 bucket0_flush_length.  Furthermore, each bucket0
    //has now received more data(logically - it's buffered) but it is not yet committed.
    //lets check that side files exist, etc
    dir = AcidUtils.getAcidState(fs, partLoc, conf, getTransactionContext(conf), null, false);
    for (AcidUtils.ParsedDelta pd : dir.getCurrentDirectories()) {
      for (FileStatus stat : fs.listStatus(pd.getPath(), AcidUtils.bucketFileFilter)) {
        Path lengthFile = OrcAcidUtils.getSideFile(stat.getPath());
        Assert.assertTrue(lengthFile + " missing", fs.exists(lengthFile));
        long lengthFileSize = fs.getFileStatus(lengthFile).getLen();
        Assert.assertTrue("Expected " + lengthFile + " to be non empty. lengh=" +
          lengthFileSize, lengthFileSize > 0);
        long logicalLength = AcidUtils.getLogicalLength(fs, stat);
        long actualLength = stat.getLen();
        Assert.assertTrue("", logicalLength <= actualLength);
      }
    }
    checkDataWritten2(partLoc, 1, 20, 2,
      validationQuery, true, "1\tHello streaming", "3\tHello streaming - once again");

    connection.commitTransaction();

    checkDataWritten2(partLoc, 1, 20, 2,
      validationQuery, false, "1\tHello streaming",
      "2\tWelcome to streaming",
      "3\tHello streaming - once again");

    connection2.commitTransaction();

    checkDataWritten2(partLoc, 1, 20, 2,
      validationQuery, true, "1\tHello streaming",
      "2\tWelcome to streaming",
      "3\tHello streaming - once again",
      "4\tWelcome to streaming - once again");

    Assert.assertEquals(HiveStreamingConnection.TxnState.COMMITTED,
        connection.getCurrentTransactionState());
    Assert.assertEquals(HiveStreamingConnection.TxnState.COMMITTED,
        connection2.getCurrentTransactionState());

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

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