@Override
  public void commitTransaction(Set<String> partitions, String key,
      String value) throws StreamingException {
    checkState();

    Set<String> createdPartitions = new HashSet<>();
    if (partitions != null) {
      for (String partition: partitions) {
        try {
          PartitionInfo info = createPartitionIfNotExists(
              Warehouse.getPartValuesFromPartName(partition));
          if (!info.isExists()) {
            createdPartitions.add(partition);
          }
        } catch (MetaException e) {
          throw new StreamingException("Partition " + partition + " is invalid.", e);
        }
      }
      connectionStats.incrementTotalPartitions(partitions.size());
    }

    currentTransactionBatch.commit(createdPartitions, key, value);
    this.partitions.addAll(
        currentTransactionBatch.getPartitions());
    connectionStats.incrementCreatedPartitions(createdPartitions.size());
    connectionStats.incrementCommittedTransactions();
  }
Returns the file that would be used to store rows under this.
parameters
@param partitionValues partition values
@param bucketId bucket id
@param minWriteId min write Id
@param maxWriteId max write Id
@param statementId statement Id
@return the location of the file.
@throws StreamingException when the path is not found
streaming/src/java/org/apache/hive/streaming/HiveStreamingConnection.java