private void commitImpl(Set<String> partitions, String key, String value)
      throws StreamingException {
    try {
      if ((key == null && value != null) || (key != null && value == null)) {
        throw new StreamingException(String.format(
            "If key is set, the value should be as well and vice versa,"
                + " key, value = %s, %s", key, value));
      }
      recordWriter.flush();
      TxnToWriteId txnToWriteId = txnToWriteIds.get(currentTxnIndex);
      if (conn.isDynamicPartitioning()) {
        List<String> partNames = new ArrayList<>(recordWriter.getPartitions());
        createdPartitions.addAll(partNames);
        if (partitions != null) {
          partNames.addAll(partitions);
        }
        if (!partNames.isEmpty()) {
          conn.getMSC().addDynamicPartitions(txnToWriteId.getTxnId(),
              txnToWriteId.getWriteId(), conn.getDatabase(),
              conn.getTable().getTableName(), partNames,
              DataOperationType.INSERT);
        }
      }

      // If it is the last transaction in the batch, then close the files and add write events.
      // We need to close the writer as file checksum can't be obtained on the opened file.
      if ((currentTxnIndex + 1) >= txnToWriteIds.size()) {
        // Replication doesn't work if txn batch size > 1 and the last txn is aborted as write events
        // are ignored by abort txn event causing data loss. However, if the last txn in the batch is
        // committed, then data gets replicated making eventually consistent, but doesn't guarantee
        // point-in-time consistency.
        recordWriter.close();

        // Add write notification events if it is enabled.
        conn.addWriteNotificationEvents();
      }

      transactionLock.lock();
      try {
        if (key != null) {
          conn.getMSC().commitTxnWithKeyValue(txnToWriteId.getTxnId(),
              tableId, key, value);
        } else {
          conn.getMSC().commitTxn(txnToWriteId.getTxnId());
        }
        // increment the min txn id so that heartbeat thread will heartbeat only from the next open transaction.
        // the current transaction is going to committed or fail, so don't need heartbeat for current transaction.
        if (currentTxnIndex + 1 < txnToWriteIds.size()) {
          minTxnId.set(txnToWriteIds.get(currentTxnIndex + 1).getTxnId());
        } else {
          // exhausted the batch, no longer have to heartbeat for current txn batch
          minTxnId.set(-1);
        }
      } finally {
        transactionLock.unlock();
      }
      state = HiveStreamingConnection.TxnState.COMMITTED;
      txnStatus[currentTxnIndex] = HiveStreamingConnection.TxnState.COMMITTED;
    } catch (NoSuchTxnException e) {
      throw new TransactionError("Invalid transaction id : "
          + getCurrentTxnId(), e);
    } catch (TxnAbortedException e) {
      throw new TransactionError("Aborted transaction "
          + "cannot be committed", e);
    } catch (TException e) {
      throw new TransactionError("Unable to commitTransaction transaction"
          + getCurrentTxnId(), e);
    }
  }
Represents a batch of transactions acquired from MetaStore.

@param conn - hive streaming connection
@throws StreamingException if failed to create new RecordUpdater for batch
streaming/src/java/org/apache/hive/streaming/TransactionBatch.java