@Test
  public void testRemainingTransactions() 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)
      .connect();
    connection.beginTransaction();
    // 1) test with txn.Commit()
    int batch = 0;
    int initialCount = connection.remainingTransactions();
    while (connection.remainingTransactions() > 0) {
      connection.beginTransaction();
      Assert.assertEquals(--initialCount, connection.remainingTransactions());
      for (int rec = 0; rec < 2; ++rec) {
        Assert.assertEquals(HiveStreamingConnection.TxnState.OPEN,
            connection.getCurrentTransactionState());
        connection.write((batch * rec + ",Hello streaming").getBytes());
      }
      connection.commitTransaction();
      Assert.assertEquals(HiveStreamingConnection.TxnState.COMMITTED,
          connection.getCurrentTransactionState());
      ++batch;
    }
    Assert.assertEquals(0, connection.remainingTransactions());
    connection.close();

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

    connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName)
      .withTable(tblName)
      .withStaticPartitionValues(partitionVals)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer)
      .withHiveConf(conf)
      .connect();
    // 2) test with txn.Abort()
    connection.beginTransaction();
    batch = 0;
    initialCount = connection.remainingTransactions();
    while (connection.remainingTransactions() > 0) {
      connection.beginTransaction();
      Assert.assertEquals(--initialCount, connection.remainingTransactions());
      for (int rec = 0; rec < 2; ++rec) {
        Assert.assertEquals(HiveStreamingConnection.TxnState.OPEN,
            connection.getCurrentTransactionState());
        connection.write((batch * rec + ",Hello streaming").getBytes());
      }
      connection.abortTransaction();
      Assert.assertEquals(HiveStreamingConnection.TxnState.ABORTED,
          connection.getCurrentTransactionState());
      ++batch;
    }
    Assert.assertEquals(0, connection.remainingTransactions());
    connection.close();

    Assert.assertEquals(HiveStreamingConnection.TxnState.INACTIVE,
        connection.getCurrentTransactionState());
  }
check that transactions that have not heartbeated and timedout get properly aborted

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