gt:       ApiServiceRef ss = args.getSourceService();
context: package com.cloudera.api.dao.impl.replication;
  
  import com.cloudera.api.ApiFeature;
  import com.cloudera.api.DataView;
  import com.cloudera.api.model.ApiHdfsCloudReplicationArguments;
  import com.cloudera.api.model.ApiHdfsReplicationArguments;
  import com.cloudera.api.model.ApiHdfsReplicationArguments.ReplicationStrategy;
  import com.cloudera.api.model.ApiHdfsReplicationCounter;
  import com.cloudera.api.model.ApiHdfsReplicationResult;
  import com.cloudera.api.model.ApiReplicationCommand;
  import com.cloudera.api.model.ApiReplicationSchedule;
  import com.cloudera.api.model.ApiReplicationScheduleDataLimits;
  import com.cloudera.api.model.ApiServiceRef;
  import com.cloudera.cmf.command.SvcCmdArgs;
  import com.cloudera.cmf.model.DbCommand;
  import com.cloudera.cmf.model.DbService;
  import com.cloudera.cmf.persist.CmfEntityManager;
  import com.cloudera.cmf.service.ReplicationUtils;
  import com.cloudera.cmf.service.ServiceDataProvider;
  import com.cloudera.cmf.service.hdfs.DistCpCloudCommand.DistCpCloudCommandArgs;
  import com.cloudera.cmf.service.hdfs.DistCpCommand;
  import com.cloudera.cmf.service.hdfs.DistCpCommand.DistCpCommandArgs;
  import com.cloudera.cmf.service.hdfs.DistCpCommand.I18nKeys;
  import com.cloudera.cmf.service.hdfs.DistCpLogFetcher;
  import com.cloudera.cmf.version.CdhReleases;
  import com.cloudera.cmf.version.Release;
  import com.cloudera.enterprise.JsonUtil2;
  import com.cloudera.enterprise.distcp.avro.CopyStatus;
  import com.cloudera.enterprise.distcp.avro.FileCopyStatus;
  import com.cloudera.server.web.cmf.CmfPath.Service;
  import com.cloudera.server.web.common.I18n;
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Objects;
  import com.google.common.base.Preconditions;
  import com.google.common.collect.Lists;
  import java.io.IOException;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Set;
  import org.apache.commons.lang.StringUtils;
  import org.joda.time.Instant;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class HdfsReplicationHandler extends ReplicationHandler
  {
    private static final Logger LOG = LoggerFactory.getLogger(HdfsReplicationHandler.class);
  
    @VisibleForTesting
    static final String MAPPER_COUNTER_GROUP = "com.cloudera.enterprise.distcp.mapred.CopyMapper$Counter";
    private static final int MAX_FILE_COPY_STATUSES_FROM_LOG = 1000;
    private static final int MAX_FAILED_FILE_COPY_STATUSES_FROM_LOG = 100;
  
    private DistCpCommand.DistCpCommandArgs deserializeArgs(String serialized) { return (DistCpCommand.DistCpCommandArgs)JsonUtil2.valueFromString(DistCpCommand.DistCpCommandArgs.class, serialized); }
  
    @VisibleForTesting
    CmfEntityManager getEm() {
      return CmfEntityManager.currentCmfEntityManager();
    }
    @VisibleForTesting
    DistCpLogFetcher getDistCpLogFetcher(ServiceDataProvider sdp, DbCommand cmd) {
      return new DistCpLogFetcher(sdp, cmd);
    }
  
    ApiHdfsReplicationArguments createApiArguments(DistCpCommand.DistCpCommandArgs args) {
      ApiServiceRef srcSvc = args.sourceService != null ? new ApiServiceRef(args.sourcePeer, args.sourceCluster, args.sourceService) : null;
  
      ApiHdfsReplicationArguments apiArgs = null;
      if ((args instanceof DistCpCloudCommand.DistCpCloudCommandArgs)) {
        DistCpCloudCommand.DistCpCloudCommandArgs cloudArgs = (DistCpCloudCommand.DistCpCloudCommandArgs)args;
        apiArgs = new ApiHdfsCloudReplicationArguments(srcSvc, cloudArgs.sourcePaths != null ? (String)cloudArgs.sourcePaths.get(0) : null, cloudArgs.destinationPath, cloudArgs.mapreduceServiceName, cloudArgs.numConcurrentMaps, cloudArgs.proxyUser, cloudArgs.sourceAccount, cloudArgs.destinationAccount);
      }
      else
      {
        apiArgs = new ApiHdfsReplicationArguments(srcSvc, args.sourcePaths != null ? (String)args.sourcePaths.get(0) : null, args.destinationPath, args.mapreduceServiceName, args.numConcurrentMaps, args.proxyUser);
      }
  
      apiArgs.setSchedulerPoolName(args.poolName);
      apiArgs.setBandwidthPerMap(args.bandwidth);
      apiArgs.setAbortOnError(!args.ignoreFailures);
      apiArgs.setRemoveMissingFiles(args.delete);
      apiArgs.setLogPath(args.log);
      apiArgs.setDryRun(args.dryRun);
      apiArgs.setSkipChecksumChecks(args.skipCrcCheck);
  
      if (!ApiFeature.DISTCP_SKIPTRASH.isAvailable()) {
        apiArgs.setSkipTrash(null);
      }
      else {
        apiArgs.setSkipTrash(Boolean.valueOf(args.skipTrash == null ? false : args.skipTrash.booleanValue()));
      }
  
      if (ApiFeature.HDFS_REPLICATION_STRATEGY.isAvailable()) {
        apiArgs.setReplicationStrategy(args.strategy == null ? ApiHdfsReplicationArguments.ReplicationStrategy.STATIC : args.strategy);
      }
  
      apiArgs.setPreserveXAttrs(Boolean.valueOf(false));
  
      if (args.preserve != null)
        for (int i = 0; i < args.preserve.length(); i++)
          switch (args.preserve.charAt(i)) {
          case 'r':
            apiArgs.setPreserveReplicationCount(true);
            break;
          case 'b':
            apiArgs.setPreserveBlockSize(true);
            break;
          case 'x':
            apiArgs.setPreserveXAttrs(Boolean.valueOf(true));
            break;
          case 'a':
          case 'g':
          case 'p':
          case 'u':
            apiArgs.setPreservePermissions(true);
          case 'c':
          case 'd':
          case 'e':
          case 'f':
          case 'h':
          case 'i':
          case 'j':
          case 'k':
          case 'l':
          case 'm':
          case 'n':
          case 'o':
          case 'q':
          case 's':
          case 't':
          case 'v':
          case 'w': }   if (!ApiFeature.DISTCP_KEEPACLXATTR.isAvailable()) {
        apiArgs.setPreserveXAttrs(null);
      }
      if (!ApiFeature.REPLICATION_HDFS_EXCLUSION_FILTERS.isAvailable())
        apiArgs.setExclusionFilters(null);
      else {
        apiArgs.setExclusionFilters(args.exclusionFilters);
      }
      return apiArgs;
    }
  
    ApiHdfsReplicationResult createReplicationResult(ServiceDataProvider sdp, DbCommand cmd, DistCpCommand.DistCpCommandArgs args, Map data, DataView view)
    {
      ApiHdfsReplicationResult res = new ApiHdfsReplicationResult();
      res.setProgress(((Integer)data.get("progress")).intValue());
      res.setSetupError((String)data.get("setupError"));
      res.setDryRun(((Boolean)data.get("dryRun")).booleanValue());
      if ((args != null) && (args.proxyUser != null) && (ApiFeature.REPLICATION_RUN_AS_USER_IN_RESULT.isAvailable()))
      {
        res.setRunAsUser(args.proxyUser);
      }
      if (ApiFeature.DISTCP_JOB_INFO.isAvailable()) {
        String jobId = (String)data.get("jobId");
        if (jobId != null) {
          res.setJobId(jobId);
  
          if (args != null) {
            DbService mr = getEm().findServiceByName(args.mapreduceServiceName);
            if (mr != null) {
              long endTime = cmd.isActive() ? System.currentTimeMillis() : cmd.getEndInstant().getMillis();
  
              String uri = CmfPath.Service.getActivityUri(mr, jobId, cmd.getStartInstant().getMillis(), endTime);
  
              res.setJobDetailsUri(uri);
            }
          }
        }
      }
  
      boolean addCounters = (!ApiFeature.REPLICATION_RESULT_VIEWS.isAvailable()) || (view == DataView.FULL);
  
      Map groups = (Map)data.get("counters");
      if (groups != null) {
        List apiCounters = Lists.newArrayList();
        long filesCopyFailed = 0L;
        long bytesCopyFailed = 0L;
  
        for (Iterator i$ = groups.entrySet().iterator(); i$.hasNext(); ) { entry = (Map.Entry)i$.next();
          mapperGroup = "com.cloudera.enterprise.distcp.mapred.CopyMapper$Counter".equals(entry.getKey());
  
          Map counters = (Map)entry.getValue();
          for (Map.Entry counter : counters.entrySet()) {
            long value = ((Number)counter.getValue()).longValue();
            if (addCounters) {
              apiCounters.add(new ApiHdfsReplicationCounter((String)entry.getKey(), (String)counter.getKey(), Long.valueOf(value)));
            }
  
            if (mapperGroup) {
              String cName = (String)counter.getKey();
              if ("FILESDRYRUN".equals(cName))
                res.setNumFilesDryRun(value);
              else if ("BYTESDRYRUN".equals(cName))
                res.setNumBytesDryRun(value);
              else if ("FILESEXPECTED".equals(cName))
                res.setNumFilesExpected(value);
              else if ("BYTESEXPECTED".equals(cName))
                res.setNumBytesExpected(value);
              else if ("COPY".equals(cName))
                res.setNumFilesCopied(value);
              else if ("BYTESCOPIED".equals(cName))
                res.setNumBytesCopied(value);
              else if ("SKIP".equals(cName))
                res.setNumFilesSkipped(value);
              else if ("BYTESSKIPPED".equals(cName))
                res.setNumBytesSkipped(value);
              else if ("FILESDELETED".equals(cName))
                res.setNumFilesDeleted(value);
              else if (("FAIL".equals(cName)) || ("SKIPONERR".equals(cName)))
                filesCopyFailed += value;
              else if (("BYTESFAILED".equals(cName)) || ("BYTESSKIPPEDONERR".equals(cName)))
              {
                bytesCopyFailed += value;
              }
            }
          }
        }
        Map.Entry entry;
        boolean mapperGroup;
        if (data.containsKey("filesExpected")) {
          res.setNumFilesExpected(((Number)data.get("filesExpected")).longValue());
        }
        if (addCounters) {
          res.setCounters(apiCounters);
        }
        res.setNumFilesCopyFailed(filesCopyFailed);
        res.setNumBytesCopyFailed(bytesCopyFailed);
      }
  
      if (ApiFeature.CDH5_SUPPORT.isAvailable()) {
        res.setSnapshottedDirs((List)data.get("snapshottedDirs"));
      }
  
      if ((ApiFeature.REPLICATION_FAILED_FILES_IN_RESULT.isAvailable()) && (!cmd.isActive())) {
        DistCpLogFetcher logFetcher = getDistCpLogFetcher(sdp, cmd);
        try {
          List statuses = logFetcher.readStatus(0L, 1000);
          List failedFiles = Lists.newLinkedList();
          for (FileCopyStatus status : statuses) {
            if (status.getStatus() == CopyStatus.ERROR) {
              failedFiles.add(status.getPath());
              if (failedFiles.size() == 100) {
                break;
              }
            }
          }
          res.setFailedFiles(failedFiles);
        }
        catch (IOException e) {
          LOG.warn("Unable to fetch copy status for command {} : {} ", new Object[] { cmd.getId(), e.getMessage() });
        }
      }
  
      return res;
    }
  
    protected SvcCmdArgs createCommandArgumentsImpl(String dbArgs, boolean forceDryRun)
    {
      DistCpCommand.DistCpCommandArgs cmdArgs = deserializeArgs(dbArgs);
      cmdArgs.dryRun |= forceDryRun;
      return cmdArgs;
    }
  
    protected SvcCmdArgs createCommandArgumentsImpl(DbService service, ApiReplicationSchedule newSchedule, String oldArguments, boolean allowDryRuns)
    {
      ApiHdfsReplicationArguments args = null;
      if (newSchedule.getHdfsCloudArguments() != null) {
        args = newSchedule.getHdfsCloudArguments();
      } else {
        args = newSchedule.getHdfsArguments();
        Preconditions.checkArgument(args != null, "Arguments not specified for HDFS replication schedule");
      }
  
      Preconditions.checkArgument((allowDryRuns) || (!args.isDryRun()), "Dry run replication schedule not allowed.");
  
      Preconditions.checkArgument(args.getMapreduceServiceName() != null, "Missing MapReduce service name.");
  
      Preconditions.checkArgument(args.getSourcePath() != null, "Missing source path.");
  
      Preconditions.checkArgument(args.getDestinationPath() != null, "Missing destination path.");
  
      Boolean preserveXAttrs = args.getPreserveXAttrs();
      if ((preserveXAttrs != null) && (preserveXAttrs.booleanValue())) {
        Preconditions.checkArgument(service.getServiceVersion().atLeast(DistCpCommand.MIN_ACLXATTRS_VERSION), String.format("XAttrs replication is only supported for CDH versions %s and higher.", new Object[] { DistCpCommand.MIN_ACLXATTRS_VERSION.getVersion() }));
      }
  
      DistCpCommand.DistCpCommandArgs oldArgs = null;
      if (oldArguments != null) {
        oldArgs = deserializeArgs(oldArguments);
  
        Preconditions.checkArgument(((String)oldArgs.sourcePaths.get(0)).equals(args.getSourcePath()), "Cannot change source path for an existing HDFS replication schedule.");
  
  