@Test
  public void testMultipleTransactionBatchCommits() 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)
      .withTransactionBatchSize(10)
      .withHiveConf(conf)
      .connect();

    connection.beginTransaction();
    connection.write("1,Hello streaming".getBytes());
    connection.commitTransaction();
    String validationQuery = "select id, msg from " + dbName + "." + tblName + " order by id, msg";
    checkDataWritten2(partLoc, 1, 10, 1, validationQuery, false, "1\tHello streaming");

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

    checkDataWritten2(partLoc, 1, 10, 1, validationQuery, true, "1\tHello streaming",
      "2\tWelcome to streaming");

    connection.close();

    connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName)
      .withTable(tblName)
      .withStaticPartitionValues(partitionVals)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer)
      .withTransactionBatchSize(10)
      .withHiveConf(conf)
      .connect();
    // 2nd Txn Batch
    connection.beginTransaction();
    connection.write("3,Hello streaming - once again".getBytes());
    connection.commitTransaction();

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

    connection.beginTransaction();
    connection.write("4,Welcome to streaming - once again".getBytes());
    connection.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());

    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