@Test
  public void testAddDynamicPartitionWithWriteId() throws Exception {
    queryTable(driver, "drop table if exists default.writeiddynamic");
    queryTable(driver, "create table default.writeiddynamic (a"
        + " string, b string) partitioned by (c string, d string)"
        + " stored as orc TBLPROPERTIES('transactional'='true')");

    StrictDelimitedInputWriter writerT =
        StrictDelimitedInputWriter.newBuilder().withFieldDelimiter(',').build();
    HiveStreamingConnection transactionConnection =
        HiveStreamingConnection.newBuilder().withDatabase("default")
            .withTable("writeiddynamic").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("writeiddynamic").withRecordWriter(writerOne)
            .withHiveConf(conf).withWriteId(writeId).withStatementId(1)
            .withTableObject(tObject).connect();

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

    Assert.assertNotNull(connectionOne);

    connectionTwo.beginTransaction();
    connectionOne.beginTransaction();
    connectionOne.write("1,2,3,4".getBytes());
    connectionOne.write("1,2,5,6".getBytes());
    connectionTwo.write("1,2,30,40".getBytes());
    connectionOne.write("1,2,7,8".getBytes());
    connectionTwo.write("1,2,50,60".getBytes());
    connectionOne.write("1,2,9,10".getBytes());
    connectionOne.commitTransaction();
    connectionTwo.commitTransaction();

    Set<String> partitionsOne = new HashSet<>(connectionOne.getPartitions());
    Assert.assertEquals(4, partitionsOne.size());

    Set<String> partitionsTwo = new HashSet<>(connectionTwo.getPartitions());
    Assert.assertEquals(2, partitionsTwo.size());

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

    try {
      String partitionName = partitionsOne.iterator().next();
      msClient.getPartition("default", "writeiddynamic", partitionName);
      Assert.fail(
          "Partition shouldn't exist so a NoSuchObjectException should have been raised");
    } catch (NoSuchObjectException e) {
    }

    partitionsOne.addAll(partitionsTwo);
    Set<String> allPartitions = partitionsOne;
    transactionConnection.commitTransaction(allPartitions);

    // Ensure partition is present
    for (String partition : allPartitions) {
      Partition p =
          msClient.getPartition("default", "writeiddynamic",
              partition);
      Assert.assertNotNull("Did not find added partition", p);
    }
  }
@param validationQuery query to read from table to compare data against {@code records}
@param records         expected data.  each row is CVS list of values
streaming/src/test/org/apache/hive/streaming/TestStreaming.java