@Test
  public void testJsonInputStream() throws Exception {
    StrictJsonWriter writer = StrictJsonWriter.newBuilder()
      .withLineDelimiterPattern("\\|")
      .build();
    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase(dbName)
      .withTable(tblName)
      .withStaticPartitionValues(partitionVals)
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(writer)
      .withHiveConf(conf)
      .connect();

    // 1st Txn
    connection.beginTransaction();
    Assert.assertEquals(HiveStreamingConnection.TxnState.OPEN, connection.getCurrentTransactionState());
    String records = "{\"id\" : 1, \"msg\": \"Hello streaming\"}|{\"id\" : 2, \"msg\": \"Hello world\"}|{\"id\" : 3, " +
      "\"msg\": \"Hello world!!\"}";
    ByteArrayInputStream bais = new ByteArrayInputStream(records.getBytes());
    connection.write(bais);
    connection.commitTransaction();
    bais.close();
    connection.close();
    List<String> rs = queryTable(driver, "select * from " + dbName + "." + tblName);
    Assert.assertEquals(3, rs.size());
    Assert.assertEquals("1\tHello streaming\tAsia\tIndia", rs.get(0));
    Assert.assertEquals("2\tHello world\tAsia\tIndia", rs.get(1));
    Assert.assertEquals("3\tHello world!!\tAsia\tIndia", rs.get(2));
  }
check that transactions that have not heartbeated and timedout get properly aborted

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