@Override
  public int execute() {
    String replPolicy = work.getReplPolicy();
    String tableName = work.getTableName();
    ReplicationSpec replicationSpec = work.getReplicationSpec();
    if ((tableName != null) && (replicationSpec != null)) {
        try {
          Database database = Hive.get().getDatabase(work.getDbName());
          if (!replicationSpec.allowReplacementInto(database.getParameters())) {
            // if the event is already replayed, then no need to replay it again.
            LOG.debug("ReplTxnTask: Event is skipped as it is already replayed. Event Id: " +
                    replicationSpec.getReplicationState() + "Event Type: " + work.getOperationType());
            return 0;
          }
        } catch (HiveException e1) {
          LOG.error("Get database failed with exception " + e1.getMessage());
          return 1;
        }
    }

    try {
      HiveTxnManager txnManager = context.getHiveTxnManager();
      String user = UserGroupInformation.getCurrentUser().getUserName();
      switch(work.getOperationType()) {
      case REPL_OPEN_TXN:
        List<Long> txnIds = txnManager.replOpenTxn(replPolicy, work.getTxnIds(), user);
        assert txnIds.size() == work.getTxnIds().size();
        LOG.info("Replayed OpenTxn Event for policy " + replPolicy + " with srcTxn " +
            work.getTxnIds().toString() + " and target txn id " + txnIds.toString());
        return 0;
      case REPL_ABORT_TXN:
        for (long txnId : work.getTxnIds()) {
          txnManager.replRollbackTxn(replPolicy, txnId);
          LOG.info("Replayed AbortTxn Event for policy " + replPolicy + " with srcTxn " + txnId);
        }
        return 0;
      case REPL_COMMIT_TXN:
        // Currently only one commit txn per event is supported.
        assert (work.getTxnIds().size() == 1);

        long txnId = work.getTxnIds().get(0);
        CommitTxnRequest commitTxnRequest = new CommitTxnRequest(txnId);
        commitTxnRequest.setReplPolicy(work.getReplPolicy());
        commitTxnRequest.setWriteEventInfos(work.getWriteEventInfos());
        commitTxnRequest.setTxn_type(TxnType.REPL_CREATED);
        txnManager.replCommitTxn(commitTxnRequest);
        LOG.info("Replayed CommitTxn Event for replPolicy: " + replPolicy + " with srcTxn: " + txnId +
            "WriteEventInfos: " + work.getWriteEventInfos());
        return 0;
      case REPL_ALLOC_WRITE_ID:
        assert work.getTxnToWriteIdList() != null;
        String dbName = work.getDbName();
        List <TxnToWriteId> txnToWriteIdList = work.getTxnToWriteIdList();

        if (ReplUtils.filterTransactionOperations(conf)) {
          // Implicitly open transactions since the OpenTxn(s) were filtered out to save space.
          List<Long> txnIdsToOpen = txnToWriteIdList.stream()
                  .map(txn2WriteId -> new Long(txn2WriteId.getTxnId()))
                  .collect(Collectors.toList());
          List<Long> openedTxnIds = txnManager.replOpenTxn(replPolicy, txnIdsToOpen, user);
          assert openedTxnIds.size() == txnIdsToOpen.size();
          LOG.info("Replayed (implicit) OpenTxn Event for policy " + replPolicy + " with srcTxn " +
                  txnIdsToOpen.toString() + " and target txn id " + openedTxnIds.toString());
        }

        txnManager.replAllocateTableWriteIdsBatch(dbName, tableName, replPolicy, txnToWriteIdList);
        LOG.info("Replayed alloc write Id Event for repl policy: " + replPolicy + " db Name : " + dbName +
            " txnToWriteIdList: " +txnToWriteIdList.toString() + " table name: " + tableName);
        return 0;
      case REPL_WRITEID_STATE:
        txnManager.replTableWriteIdState(work.getValidWriteIdList(), work.getDbName(), tableName, work.getPartNames());
        LOG.info("Replicated WriteId state for DbName: " + work.getDbName() +
            " TableName: " + tableName +
            " ValidWriteIdList: " + work.getValidWriteIdList());
        return 0;
      default:
        LOG.error("Operation Type " + work.getOperationType() + " is not supported ");
        return 1;
      }
    } catch (Exception e) {
      console.printError("Failed with exception " + e.getMessage(), "\n"
          + StringUtils.stringifyException(e));
      setException(e);
      LOG.error("ReplTxnTask failed", e);
      return ReplUtils.handleException(true, e, work.getDumpDirectory(), work.getMetricCollector(),
              getName(), conf);
    }
  }
ReplTxnTask.
Used for replaying the transaction related events.
ql/src/java/org/apache/hadoop/hive/ql/exec/ReplTxnTask.java