@Test
  public void testTransactionBatchAbortAndCommit() throws Exception {
    String agentInfo = "UT_" + Thread.currentThread().getName();
    StrictDelimitedInputWriter writer = StrictDelimitedInputWriter.newBuilder()
      .withFieldDelimiter(',')
      .build();

    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName)
      .withTable(tblName)
      .withStaticPartitionValues(partitionVals)
      .withAgentInfo(agentInfo)
      .withRecordWriter(writer)
      .withHiveConf(conf)
      .withTransactionBatchSize(10)
      .connect();

    connection.beginTransaction();
    connection.write("1,Hello streaming".getBytes());
    connection.write("2,Welcome to streaming".getBytes());
    ShowLocksResponse resp = msClient.showLocks(new ShowLocksRequest());
    Assert.assertEquals("LockCount", 1, resp.getLocksSize());
    Assert.assertEquals("LockType", LockType.SHARED_READ, resp.getLocks().get(0).getType());
    Assert.assertEquals("LockState", LockState.ACQUIRED, resp.getLocks().get(0).getState());
    Assert.assertEquals("AgentInfo", agentInfo, resp.getLocks().get(0).getAgentInfo());
    connection.abortTransaction();

    checkNothingWritten(partLoc);

    Assert.assertEquals(HiveStreamingConnection.TxnState.ABORTED,
        connection.getCurrentTransactionState());

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

    checkDataWritten(partLoc, 1, 10, 1, 1, "{1, Hello streaming}",
      "{2, Welcome to streaming}");

    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