private void createReplLoadCompleteAckTask() {
    if (!work.hasBootstrapLoadTasks()
            && (work.isIncrementalLoad() ? !work.incrementalLoadTasksBuilder().hasMoreWork() : true)) {
      //All repl load tasks are executed and status is 0, create the task to add the acknowledgement
      List<PreAckTask> listOfPreAckTasks = new LinkedList<>();
      listOfPreAckTasks.add(new PreAckTask() {
        @Override
        public void run() throws SemanticException {
          try {
            IMetaStoreClient client = getHive().getMSC();
            long currentNotificationID = client.getCurrentNotificationEventId().getEventId();
            Path loadMetadataFilePath = new Path(work.dumpDirectory, LOAD_METADATA.toString());
            Utils.writeOutput(String.valueOf(currentNotificationID), loadMetadataFilePath, conf);
            LOG.info("Created LOAD Metadata file : {} with NotificationID : {}",
                    loadMetadataFilePath, currentNotificationID);
          } catch (TException ex) {
            throw new SemanticException(ex);
          }
        }
      });
      if (work.shouldFailover()) {
        listOfPreAckTasks.add(new PreAckTask() {
          @Override
          public void run() throws SemanticException {
            try {
              Database db = getHive().getDatabase(work.getTargetDatabase());
              if (MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(db, MetaStoreUtils.FailoverEndpoint.TARGET)) {
                return;
              }
              HashMap<String, String> params = new HashMap<>(db.getParameters());

              LOG.info("Setting failover endpoint:{} to TARGET for database: {}", ReplConst.REPL_FAILOVER_ENDPOINT, db.getName());
              params.put(ReplConst.REPL_FAILOVER_ENDPOINT, MetaStoreUtils.FailoverEndpoint.TARGET.toString());
              LOG.info("Setting {} for database: {}", ReplConst.REPL_ENABLE_BACKGROUND_THREAD, db.getName());
              params.put(ReplConst.REPL_ENABLE_BACKGROUND_THREAD,"true");

              // we are here due to planned failover
              params.put(ReplConst.REPL_METRICS_LAST_FAILOVER_TYPE, ReplConst.FailoverType.PLANNED.toString());
              LOG.info("Replication Metrics: Setting last failover type for database: {} to: {} ", db.getName(), ReplConst.FailoverType.PLANNED.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: {} ", db.getName(), failoverCount);
              params.put(ReplConst.REPL_METRICS_FAILOVER_COUNT, Integer.toString(failoverCount));

              db.setParameters(params);
              getHive().alterDatabase(work.getTargetDatabase(), db);
            } catch (HiveException e) {
              throw new SemanticException(e);
            }
          }
        });
      }
      if(work.isSecondFailover) {
        // If it is the second load of optimised bootstrap that means this is the end of the cycle, add tasks to sort
        // out the database properties.
        listOfPreAckTasks.add(new PreAckTask() {
          @Override
          public void run() throws SemanticException {
            try {
              Hive hiveDb = getHive();
              Database db = hiveDb.getDatabase(work.getTargetDatabase());
              HashMap<String, String> params = new HashMap<>(db.getParameters());
              LOG.debug("Database {} properties before removal {}", work.getTargetDatabase(), params);
              params.remove(REPL_RESUME_STARTED_AFTER_FAILOVER);
              params.remove(SOURCE_OF_REPLICATION);
              if (!work.shouldFailover()) {
                params.remove(REPL_FAILOVER_ENDPOINT);
              }

              LOG.info("Removed {} property from database {} after successful optimised bootstrap load.",
                  SOURCE_OF_REPLICATION, work.getTargetDatabase());

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

              long failbackEndTime = System.currentTimeMillis();
              params.put(ReplConst.REPL_METRICS_LAST_FAILBACK_ENDTIME, Long.toString(failbackEndTime));
              LOG.info("Replication Metrics: Setting replication metrics failback end time for database: {} to: {} ", db.getName(), failbackEndTime);

              db.setParameters(params);
              hiveDb.alterDatabase(work.getTargetDatabase(), db);
              LOG.debug("Database {} properties after removal {}", work.getTargetDatabase(), params);
            } catch (HiveException e) {
              throw new SemanticException(e);
            }
          }
        });
      }
      AckWork replLoadAckWork = new AckWork(new Path(work.dumpDirectory, LOAD_ACKNOWLEDGEMENT.toString()),
              work.getMetricCollector(), listOfPreAckTasks);
      Task<AckWork> loadAckWorkTask = TaskFactory.get(replLoadAckWork, conf);
      if (childTasks.isEmpty()) {
        childTasks.add(loadAckWorkTask);
      } else {
        DAGTraversal.traverse(childTasks, new AddDependencyToLeaves(Collections.singletonList(loadAckWorkTask)));
      }
    }
  }
If replication policy is changed between previous and current load, then the excluded tables in
the new replication policy will be dropped.

@throws HiveException Failed to get/drop the tables.
ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ReplLoadTask.java