@Test
  public void testConnectionWithWriteId() throws Exception {
    queryTable(driver, "drop table if exists default.writeidconnection");
    queryTable(driver, "create table default.writeidconnection (a string, b string) stored as orc " +
        "TBLPROPERTIES('transactional'='true')");
    queryTable(driver, "insert into default.writeidconnection values('a0','bar')");

    List<String> rs = queryTable(driver, "select * from default.writeidconnection");
    Assert.assertEquals(1, rs.size());
    Assert.assertEquals("a0\tbar", rs.get(0));

    StrictDelimitedInputWriter writerT = StrictDelimitedInputWriter.newBuilder()
        .withFieldDelimiter(',')
        .build();
    HiveStreamingConnection transactionConnection = HiveStreamingConnection.newBuilder()
        .withDatabase("Default")
        .withTable("writeidconnection")
        .withRecordWriter(writerT)
        .withHiveConf(conf)
        .connect();
    transactionConnection.beginTransaction();

    Table tObject = transactionConnection.getTable();
    Long writeId = transactionConnection.getCurrentWriteId();

    Assert.assertNotNull(tObject);
    Assert.assertNotNull(writeId);

    StrictDelimitedInputWriter writerOne = StrictDelimitedInputWriter.newBuilder()
        .withFieldDelimiter(',')
        .build();
    HiveStreamingConnection connectionOne = HiveStreamingConnection.newBuilder()
        .withDatabase("Default")
        .withTable("writeidconnection")
        .withRecordWriter(writerOne)
        .withHiveConf(conf)
        .withWriteId(writeId)
        .withStatementId(1)
        .withTableObject(tObject)
        .connect();

    StrictDelimitedInputWriter writerTwo = StrictDelimitedInputWriter.newBuilder()
        .withFieldDelimiter(',')
        .build();
    HiveStreamingConnection connectionTwo = HiveStreamingConnection.newBuilder()
        .withDatabase("Default")
        .withRecordWriter(writerTwo)
        .withHiveConf(conf)
        .withWriteId(writeId)
        .withStatementId(2)
        .withTableObject(tObject)
        .connect();

    Assert.assertNotNull(connectionOne);
    Assert.assertNotNull(connectionTwo);

    connectionOne.beginTransaction();
    connectionTwo.beginTransaction();
    connectionOne.write("a1,b2".getBytes());
    connectionTwo.write("a5,b6".getBytes());
    connectionOne.write("a3,b4".getBytes());
    connectionOne.commitTransaction();
    connectionTwo.commitTransaction();

    Assert.assertEquals(HiveStreamingConnection.TxnState.PREPARED_FOR_COMMIT,
        connectionOne.getCurrentTransactionState());
    Assert.assertEquals(HiveStreamingConnection.TxnState.PREPARED_FOR_COMMIT,
        connectionTwo.getCurrentTransactionState());

    try {
      connectionOne.beginTransaction();
      Assert.fail("second beginTransaction should have thrown a "
          + "StreamingException");
    } catch (StreamingException e) {

    }

    connectionOne.close();
    connectionTwo.close();

    rs = queryTable(driver, "select ROW__ID, a, b, "
        + "INPUT__FILE__NAME from default.writeidconnection order by ROW__ID");
    // Nothing here since it hasn't been committed
    Assert.assertEquals(1, rs.size());

    transactionConnection.commitTransaction();

    rs = queryTable(driver, "select ROW__ID, a, b, "
        + "INPUT__FILE__NAME from default.writeidconnection order by a");
    Assert.assertEquals(4, rs.size());
    Assert.assertTrue(rs.get(0), rs.get(0).startsWith("{\"writeid\":1,\"bucketid\":536870912,\"rowid\":0}\ta0\tbar"));
    Assert.assertTrue(rs.get(0), rs.get(0).endsWith("bucket_00000_0"));
    Assert.assertTrue(rs.get(1), rs.get(1).contains("\"rowid\":0}\ta1\tb2"));
    Assert.assertTrue(rs.get(1), rs.get(1).endsWith("bucket_00000"));
    Assert.assertTrue(rs.get(2), rs.get(2).contains("\"rowid\":1}\ta3\tb4"));
    Assert.assertTrue(rs.get(2), rs.get(2).endsWith("bucket_00000"));
    Assert.assertTrue(rs.get(3), rs.get(3).contains("\ta5\tb6"));
    Assert.assertTrue(rs.get(3), rs.get(3).endsWith("bucket_00000"));
  }
Test that streaming can write to unbucketed table.
streaming/src/test/org/apache/hive/streaming/TestStreaming.java