private int executeIncrementalLoad(long loadStartTime) throws Exception {
    // If replication policy is changed between previous and current repl load, then drop the tables
    // that are excluded in the new replication policy.
    if (work.replScopeModified) {
      dropTablesExcludedInReplScope(work.currentReplScope);
    }
    Database targetDb = getHive().getDatabase(work.dbNameToLoadIn);
    Map<String, String> props = new HashMap<>();
    // check if db is set READ_ONLY, if not then set it. Basically this ensures backward
    // compatibility.
    if (!isDbReadOnly(targetDb) && isReadOnlyHookRegistered()) {
      setDbReadOnly();
    }
    // Check if it is a optimised bootstrap failover.
    if (work.isFirstFailover) {
      // Check it should be marked as target of replication & not source of replication.
      if (MetaStoreUtils.isTargetOfReplication(targetDb)) {
        LOG.error("The database {} is already marked as target for replication", targetDb.getName());
        throw new Exception("Failover target is already marked as target");
      }
      if (!ReplChangeManager.isSourceOfReplication(targetDb)) {
        LOG.error("The database {} is already source of replication.", targetDb.getName());
        throw new Exception("Failover target was not source of replication");
      }
      Map<String, String> params = targetDb.getParameters();
      if (params == null) {
        params = new HashMap<>();
      }

      long failbackStartTime = System.currentTimeMillis();
      params.put(ReplConst.REPL_METRICS_LAST_FAILBACK_STARTTIME, Long.toString(failbackStartTime));
      LOG.info("Replication Metrics: Setting replication metrics failback starttime for database: {} to: {} ", targetDb.getName(), failbackStartTime);
      if (!MetaStoreUtils.isDbBeingPlannedFailedOver(targetDb)) { // if this is failback due to unplanned failover
        LOG.info("Replication Metrics: Setting last failover type for database: {} to: {} ", targetDb.getName(), ReplConst.FailoverType.UNPLANNED.toString());
        params.put(ReplConst.REPL_METRICS_LAST_FAILOVER_TYPE, ReplConst.FailoverType.UNPLANNED.toString());

        int failoverCount = 1 + NumberUtils.toInt(params.getOrDefault(ReplConst.REPL_METRICS_FAILOVER_COUNT, "0"), 0);
        LOG.info("Replication Metrics: Setting replication metrics failover count for database: {} to: {} ", targetDb.getName(), failoverCount);
        params.put(ReplConst.REPL_METRICS_FAILOVER_COUNT, Integer.toString(failoverCount));
      }

      work.getMetricCollector().reportStageStart(STAGE_NAME, new HashMap<>());
      boolean isTableDiffPresent =
          checkFileExists(new Path(work.dumpDirectory).getParent(), conf, TABLE_DIFF_COMPLETE_DIRECTORY);
      boolean isAbortTxnsListPresent =
              checkFileExists(new Path(work.dumpDirectory).getParent(), conf, OptimisedBootstrapUtils.ABORT_TXNS_FILE);
      Long eventId = Long.parseLong(getEventIdFromFile(new Path(work.dumpDirectory).getParent(), conf)[0]);
      List<NotificationEvent> notificationEvents = OptimisedBootstrapUtils.getListOfNotificationEvents(eventId, getHive(), work);
      if (!isAbortTxnsListPresent) {
        //Abort the ongoing transactions(opened prior to failover) for the target database.
        HiveTxnManager hiveTxnManager = getTxnMgr();
        ValidTxnList validTxnList = hiveTxnManager.getValidTxns(excludedTxns);
        Set<Long> allOpenTxns = new HashSet<>(ReplUtils.getOpenTxns(validTxnList));
        abortOpenTxnsForDatabase(hiveTxnManager, validTxnList, work.dbNameToLoadIn, getHive());
        //Re-fetch the list of notification events post failover eventId.
        notificationEvents = OptimisedBootstrapUtils.getListOfNotificationEvents(eventId, getHive(), work);
        OptimisedBootstrapUtils.prepareAbortTxnsFile(notificationEvents, allOpenTxns,
                new Path(work.dumpDirectory).getParent(), conf);
      }
      if (!isTableDiffPresent) {
        prepareTableDiffFile(notificationEvents, getHive(), work, conf);
      }
      if (this.childTasks == null) {
        this.childTasks = new ArrayList<>();
      }
      createReplLoadCompleteAckTask();
      work.getMetricCollector().reportStageEnd(STAGE_NAME, Status.SUCCESS);
      if (Boolean.parseBoolean(targetDb.getParameters().get(REPL_RESUME_STARTED_AFTER_FAILOVER))) {
        work.getMetricCollector().reportEnd(Status.RESUME_READY);
      } else {
        work.getMetricCollector().reportEnd(Status.SUCCESS);
      }
      targetDb.setParameters(params);
      getHive().alterDatabase(work.dbNameToLoadIn, targetDb);
      return 0;
    } else if (work.isSecondFailover) {
      // DROP the tables extra on target, which are not on source cluster.

      Hive db = getHive();
      for (String table : work.tablesToDrop) {
        LOG.info("Dropping table {} for optimised bootstrap", work.dbNameToLoadIn + "." + table);
        db.dropTable(work.dbNameToLoadIn + "." + table, true);
      }
      Database sourceDb = getSourceDbMetadata();  //This sourceDb was the actual target prior to failover.
      Map<String, String> sourceDbProps = sourceDb.getParameters();
      Map<String, String> targetDbProps = new HashMap<>(targetDb.getParameters());
      for (String key : MetaStoreUtils.getReplicationDbProps()) {
        //Replication Props will be handled separately as part of preAckTask.
        targetDbProps.remove(key);
      }
      for (Map.Entry<String, String> currProp : targetDbProps.entrySet()) {
        String actualVal = sourceDbProps.get(currProp.getKey());
        if (!currProp.getValue().equals(actualVal)) {
          props.put(currProp.getKey(), (actualVal == null) ? "" : actualVal);
        }
      }
      AlterDatabaseSetOwnerDesc alterDbDesc = new AlterDatabaseSetOwnerDesc(sourceDb.getName(),
              new PrincipalDesc(sourceDb.getOwnerName(), sourceDb.getOwnerType()), null);
      DDLWork ddlWork = new DDLWork(new HashSet<>(), new HashSet<>(), alterDbDesc, true,
              (new Path(work.dumpDirectory)).getParent().toString(), work.getMetricCollector());
      if (this.childTasks == null) {
        this.childTasks = new ArrayList<>();
      }
      this.childTasks.add(TaskFactory.get(ddlWork, conf));
    }
    if (!MetaStoreUtils.isTargetOfReplication(targetDb)) {
      props.put(ReplConst.TARGET_OF_REPLICATION, ReplConst.TRUE);
    }
    if (!work.shouldFailover() && MetaStoreUtils.isDbBeingPlannedFailedOver(targetDb)) {
      props.put(ReplConst.REPL_FAILOVER_ENDPOINT, "");
    }
    if (!props.isEmpty()) {
      AlterDatabaseSetPropertiesDesc setTargetDesc = new AlterDatabaseSetPropertiesDesc(work.dbNameToLoadIn, props, null);
      Task<?> addReplTargetPropTask =
              TaskFactory.get(new DDLWork(new HashSet<>(), new HashSet<>(), setTargetDesc, true,
                      work.dumpDirectory, work.getMetricCollector()), conf);
      if (this.childTasks == null) {
        this.childTasks = new ArrayList<>();
      }
      this.childTasks.add(addReplTargetPropTask);
    }
    IncrementalLoadTasksBuilder builder = work.incrementalLoadTasksBuilder();
    // If incremental events are already applied, then check and perform if need to bootstrap any tables.
    if (!builder.hasMoreWork() && work.isLastReplIDUpdated()) {
      if (work.hasBootstrapLoadTasks()) {
        LOG.debug("Current incremental dump have tables to be bootstrapped. Switching to bootstrap "
            + "mode after applying all events.");
        return executeBootStrapLoad();
      }
    }
    List<Task<?>> childTasks = new ArrayList<>();
    int maxTasks = conf.getIntVar(HiveConf.ConfVars.REPL_APPROX_MAX_LOAD_TASKS);
    TaskTracker tracker = new TaskTracker(maxTasks);
    addLazyDataCopyTask(tracker, builder.getReplLogger());
    childTasks.add(builder.build(context, getHive(), LOG, tracker));
    // If there are no more events to be applied, add a task to update the last.repl.id of the
    // target database to the event id of the last event considered by the dump. Next
    // incremental cycle won't consider the events in this dump again if it starts from this id.
    if (!builder.hasMoreWork()) {
      // The name of the database to be loaded into is either specified directly in REPL LOAD
      // command i.e. when dbNameToLoadIn has a valid dbname or is available through dump
      // metadata during table level replication.
      String dbName = work.dbNameToLoadIn;
      if (dbName == null || StringUtils.isBlank(dbName)) {
        if (work.currentReplScope != null) {
          String replScopeDbName = work.currentReplScope.getDbName();
          if (replScopeDbName != null && !"*".equals(replScopeDbName)) {
            dbName = replScopeDbName;
          }
        }
      }
      // If we are replicating to multiple databases at a time, it's not
      // possible to know which all databases we are replicating into and hence we can not
      // update repl id in all those databases.
      if (StringUtils.isNotBlank(dbName)) {
        String lastEventid = builder.eventTo().toString();
        Map<String, String> mapProp = new HashMap<>();
        mapProp.put(ReplicationSpec.KEY.CURR_STATE_ID_SOURCE.toString(), lastEventid);
        AlterDatabaseSetPropertiesDesc alterDbDesc =
            new AlterDatabaseSetPropertiesDesc(dbName, mapProp,
                new ReplicationSpec(lastEventid, lastEventid));
        Task<?> updateReplIdTask =
            TaskFactory.get(new DDLWork(new HashSet<>(), new HashSet<>(), alterDbDesc, true,
                    (new Path(work.dumpDirectory).getParent()).toString(), work.getMetricCollector()), conf);
        DAGTraversal.traverse(childTasks, new AddDependencyToLeaves(updateReplIdTask));
        work.setLastReplIDUpdated(true);
        LOG.debug("Added task to set last repl id of db " + dbName + " to " + lastEventid);
      }
    }
    // Once all the incremental events are applied, enable bootstrap of tables if exist.
    if (builder.hasMoreWork() || work.hasBootstrapLoadTasks()) {
      DAGTraversal.traverse(childTasks, new AddDependencyToLeaves(TaskFactory.get(work, conf)));
    }
    if (this.childTasks == null) {
      this.childTasks = new ArrayList<>();
    }
    this.childTasks.addAll(childTasks);
    createReplLoadCompleteAckTask();
    // Clean-up snapshots
    if (conf.getBoolVar(REPL_SNAPSHOT_DIFF_FOR_EXTERNAL_TABLE_COPY)) {
      cleanupSnapshots(new Path(work.getDumpDirectory()).getParent().getParent().getParent(),
          work.getSourceDbName().toLowerCase(), conf, null, true);
    }

    //pass the current time at the end of repl-load stage as the starting time of the first event.
    long currentTimestamp = System.currentTimeMillis();
    ((IncrementalLoadLogger)work.incrementalLoadTasksBuilder().getReplLogger()).initiateEventTimestamp(currentTimestamp);
    LOG.info("REPL_INCREMENTAL_LOAD stage duration : {} ms", currentTimestamp - loadStartTime);
    return 0;
  }
There was a database update done before and we want to make sure we update the last repl
id on this database as we are now going to switch to processing a new database.
This has to be last task in the graph since if there are intermediate tasks and the last.repl.id
is a root level task then in the execution phase the root level tasks will get executed first,
however if any of the child tasks of the bootstrap load failed then even though the bootstrap has failed
the last repl status of the target database will return a valid value, which will not represent
the state of the database.
ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ReplLoadTask.java