alluxio-master-2.x/core/server/master/src/main/java/alluxio/master/file/FileSystemMasterClientServiceHandler.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.master.file;

import alluxio.AlluxioURI;
import alluxio.RpcUtils;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.FileDoesNotExistException;
import alluxio.grpc.CancelSyncMetadataPRequest;
import alluxio.grpc.CancelSyncMetadataPResponse;
import alluxio.grpc.CheckAccessPRequest;
import alluxio.grpc.CheckAccessPResponse;
import alluxio.grpc.CheckConsistencyPOptions;
import alluxio.grpc.CheckConsistencyPRequest;
import alluxio.grpc.CheckConsistencyPResponse;
import alluxio.grpc.CompleteFilePRequest;
import alluxio.grpc.CompleteFilePResponse;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateDirectoryPRequest;
import alluxio.grpc.CreateDirectoryPResponse;
import alluxio.grpc.CreateFilePRequest;
import alluxio.grpc.CreateFilePResponse;
import alluxio.grpc.DeletePRequest;
import alluxio.grpc.DeletePResponse;
import alluxio.grpc.ExistsPOptions;
import alluxio.grpc.ExistsPRequest;
import alluxio.grpc.ExistsPResponse;
import alluxio.grpc.FileSystemMasterClientServiceGrpc;
import alluxio.grpc.FreePRequest;
import alluxio.grpc.FreePResponse;
import alluxio.grpc.GetFilePathPRequest;
import alluxio.grpc.GetFilePathPResponse;
import alluxio.grpc.GetJobProgressPRequest;
import alluxio.grpc.GetJobProgressPResponse;
import alluxio.grpc.GetMountTablePRequest;
import alluxio.grpc.GetMountTablePResponse;
import alluxio.grpc.GetNewBlockIdForFilePRequest;
import alluxio.grpc.GetNewBlockIdForFilePResponse;
import alluxio.grpc.GetStateLockHoldersPRequest;
import alluxio.grpc.GetStateLockHoldersPResponse;
import alluxio.grpc.GetStatusPOptions;
import alluxio.grpc.GetStatusPRequest;
import alluxio.grpc.GetStatusPResponse;
import alluxio.grpc.GetSyncPathListPRequest;
import alluxio.grpc.GetSyncPathListPResponse;
import alluxio.grpc.GetSyncProgressPRequest;
import alluxio.grpc.GetSyncProgressPResponse;
import alluxio.grpc.GrpcUtils;
import alluxio.grpc.JobProgressReportFormat;
import alluxio.grpc.ListStatusPRequest;
import alluxio.grpc.ListStatusPResponse;
import alluxio.grpc.ListStatusPartialPRequest;
import alluxio.grpc.ListStatusPartialPResponse;
import alluxio.grpc.MountPRequest;
import alluxio.grpc.MountPResponse;
import alluxio.grpc.NeedsSyncRequest;
import alluxio.grpc.NeedsSyncResponse;
import alluxio.grpc.RenamePRequest;
import alluxio.grpc.RenamePResponse;
import alluxio.grpc.ReverseResolvePRequest;
import alluxio.grpc.ReverseResolvePResponse;
import alluxio.grpc.ScheduleAsyncPersistencePRequest;
import alluxio.grpc.ScheduleAsyncPersistencePResponse;
import alluxio.grpc.SetAclPRequest;
import alluxio.grpc.SetAclPResponse;
import alluxio.grpc.SetAttributePRequest;
import alluxio.grpc.SetAttributePResponse;
import alluxio.grpc.StartSyncPRequest;
import alluxio.grpc.StartSyncPResponse;
import alluxio.grpc.StopJobPRequest;
import alluxio.grpc.StopJobPResponse;
import alluxio.grpc.StopSyncPRequest;
import alluxio.grpc.StopSyncPResponse;
import alluxio.grpc.SubmitJobPRequest;
import alluxio.grpc.SubmitJobPResponse;
import alluxio.grpc.SyncMetadataAsyncPResponse;
import alluxio.grpc.SyncMetadataPRequest;
import alluxio.grpc.SyncMetadataPResponse;
import alluxio.grpc.UnmountPRequest;
import alluxio.grpc.UnmountPResponse;
import alluxio.grpc.UpdateMountPRequest;
import alluxio.grpc.UpdateMountPResponse;
import alluxio.grpc.UpdateUfsModePRequest;
import alluxio.grpc.UpdateUfsModePResponse;
import alluxio.job.JobDescription;
import alluxio.job.JobRequest;
import alluxio.job.util.SerializationUtils;
import alluxio.master.file.contexts.CheckAccessContext;
import alluxio.master.file.contexts.CheckConsistencyContext;
import alluxio.master.file.contexts.CompleteFileContext;
import alluxio.master.file.contexts.CreateDirectoryContext;
import alluxio.master.file.contexts.CreateFileContext;
import alluxio.master.file.contexts.DeleteContext;
import alluxio.master.file.contexts.ExistsContext;
import alluxio.master.file.contexts.FreeContext;
import alluxio.master.file.contexts.GetStatusContext;
import alluxio.master.file.contexts.GrpcCallTracker;
import alluxio.master.file.contexts.ListStatusContext;
import alluxio.master.file.contexts.MountContext;
import alluxio.master.file.contexts.RenameContext;
import alluxio.master.file.contexts.ScheduleAsyncPersistenceContext;
import alluxio.master.file.contexts.SetAclContext;
import alluxio.master.file.contexts.SetAttributeContext;
import alluxio.master.file.contexts.SyncMetadataContext;
import alluxio.master.job.JobFactoryProducer;
import alluxio.master.scheduler.Scheduler;
import alluxio.recorder.Recorder;
import alluxio.scheduler.job.Job;
import alluxio.underfs.UfsMode;
import alluxio.util.io.PathUtils;
import alluxio.wire.MountPointInfo;
import alluxio.wire.SyncPointInfo;

import com.google.common.base.Preconditions;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * This class is a gRPC handler for file system master RPCs invoked by an Alluxio client.
 */
public final class FileSystemMasterClientServiceHandler
    extends FileSystemMasterClientServiceGrpc.FileSystemMasterClientServiceImplBase {
  private static final Logger LOG =
      LoggerFactory.getLogger(FileSystemMasterClientServiceHandler.class);
  private final FileSystemMaster mFileSystemMaster;
  private final Scheduler mScheduler;

  /**
   * Creates a new instance of {@link FileSystemMasterClientServiceHandler}.
   *
   * @param fileSystemMaster the {@link FileSystemMaster} the handler uses internally
   * @param scheduler the {@link Scheduler}
   */
  public FileSystemMasterClientServiceHandler(FileSystemMaster fileSystemMaster,
      Scheduler scheduler) {
    Preconditions.checkNotNull(fileSystemMaster, "fileSystemMaster");
    mFileSystemMaster = fileSystemMaster;
    mScheduler = Preconditions.checkNotNull(scheduler, "scheduler");
  }

  @Override
  public void checkAccess(CheckAccessPRequest request,
      StreamObserver<CheckAccessPResponse> responseObserver) {
    RpcUtils.call(LOG,
        () -> {
          AlluxioURI pathUri = getAlluxioURI(request.getPath());
          mFileSystemMaster.checkAccess(pathUri,
              CheckAccessContext.create(request.getOptions().toBuilder()));
          return CheckAccessPResponse.getDefaultInstance();
        }, "CheckAccess", "request=%s", responseObserver, request);
  }

  @Override
  public void checkConsistency(CheckConsistencyPRequest request,
      StreamObserver<CheckConsistencyPResponse> responseObserver) {
    CheckConsistencyPOptions options = request.getOptions();
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      List<AlluxioURI> inconsistentUris = mFileSystemMaster.checkConsistency(pathUri,
          CheckConsistencyContext.create(options.toBuilder()));
      List<String> uris = new ArrayList<>(inconsistentUris.size());
      for (AlluxioURI uri : inconsistentUris) {
        uris.add(uri.getPath());
      }
      return CheckConsistencyPResponse.newBuilder().addAllInconsistentPaths(uris).build();
    }, "CheckConsistency", "request=%s", responseObserver, request);
  }

  @Override
  public void exists(ExistsPRequest request,
          StreamObserver<ExistsPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      boolean exists = mFileSystemMaster.exists(pathUri,
          ExistsContext.create(request.getOptions().toBuilder()));
      return ExistsPResponse.newBuilder().setExists(exists).build();
    }, "CheckExistence", "request=%s", responseObserver, request);
  }

  @Override
  public void completeFile(CompleteFilePRequest request,
      StreamObserver<CompleteFilePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      mFileSystemMaster.completeFile(pathUri,
          CompleteFileContext.create(request.getOptions().toBuilder()));
      return CompleteFilePResponse.newBuilder().build();
    }, "CompleteFile", "request=%s", responseObserver, request);
  }

  private void checkBucketPathExists(String path)
      throws AlluxioException, IOException {

    String bucketPath = PathUtils.getFirstLevelDirectory(path);
    boolean exists = mFileSystemMaster.exists(getAlluxioURI(bucketPath),
        ExistsContext.create(ExistsPOptions.getDefaultInstance().toBuilder()));
    if (!exists) {
      throw new FileDoesNotExistException(
          ExceptionMessage.BUCKET_DOES_NOT_EXIST.getMessage(bucketPath));
    }
  }

  @Override
  public void createDirectory(CreateDirectoryPRequest request,
      StreamObserver<CreateDirectoryPResponse> responseObserver) {
    CreateDirectoryPOptions options = request.getOptions();
    RpcUtils.call(LOG, () -> {
      if (request.getOptions().getCheckS3BucketPath()) {
        checkBucketPathExists(request.getPath());
      }
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      mFileSystemMaster.createDirectory(pathUri, CreateDirectoryContext.create(options.toBuilder())
          .withTracker(new GrpcCallTracker(responseObserver)));
      return CreateDirectoryPResponse.newBuilder().build();
    }, "CreateDirectory", "request=%s", responseObserver, request);
  }

  @Override
  public void createFile(CreateFilePRequest request,
      StreamObserver<CreateFilePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      if (request.getOptions().getCheckS3BucketPath()) {
        checkBucketPathExists(request.getPath());
      }
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      return CreateFilePResponse.newBuilder()
          .setFileInfo(GrpcUtils.toProto(mFileSystemMaster.createFile(pathUri,
              CreateFileContext.create(request.getOptions().toBuilder())
                  .withTracker(new GrpcCallTracker(responseObserver)))))
          .build();
    }, "CreateFile", "request=%s", responseObserver, request);
  }

  @Override
  public void free(FreePRequest request, StreamObserver<FreePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      mFileSystemMaster.free(pathUri, FreeContext.create(request.getOptions().toBuilder()));
      return FreePResponse.newBuilder().build();
    }, "Free", "request=%s", responseObserver, request);
  }

  @Override
  public void getNewBlockIdForFile(GetNewBlockIdForFilePRequest request,
      StreamObserver<GetNewBlockIdForFilePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      return GetNewBlockIdForFilePResponse.newBuilder()
          .setId(mFileSystemMaster.getNewBlockIdForFile(pathUri)).build();
    }, "GetNewBlockIdForFile", "request=%s", responseObserver, request);
  }

  @Override
  public void getFilePath(GetFilePathPRequest request,
      StreamObserver<GetFilePathPResponse> responseObserver) {
    long fileId = request.getFileId();
    RpcUtils.call(LOG,
        () -> GetFilePathPResponse.newBuilder()
            .setPath(mFileSystemMaster.getPath(fileId).toString()).build(),
        "GetFilePath", true, "request=%s", responseObserver, request);
  }

  @Override
  public void getStatus(GetStatusPRequest request,
      StreamObserver<GetStatusPResponse> responseObserver) {
    GetStatusPOptions options = request.getOptions();
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      return GetStatusPResponse.newBuilder()
          .setFileInfo(GrpcUtils.toProto(mFileSystemMaster.getFileInfo(pathUri, GetStatusContext
              .create(options.toBuilder()).withTracker(new GrpcCallTracker(responseObserver)))))
          .build();
    }, "GetStatus", true, "request=%s", responseObserver, request);
  }

  @Override
  public void listStatus(ListStatusPRequest request,
      StreamObserver<ListStatusPResponse> responseObserver) {
    final int listStatusBatchSize =
        Configuration.getInt(PropertyKey.MASTER_FILE_SYSTEM_LISTSTATUS_RESULTS_PER_MESSAGE);

    // Result streamer for listStatus.
    ListStatusResultStream resultStream =
        new ListStatusResultStream(listStatusBatchSize, responseObserver);

    try {
      RpcUtils.callAndReturn(LOG, () -> {
        AlluxioURI pathUri = getAlluxioURI(request.getPath());
        mFileSystemMaster.listStatus(pathUri,
            ListStatusContext.create(request.getOptions().toBuilder())
                .withTracker(new GrpcCallTracker(responseObserver)),
            resultStream);
        // Return just something.
        return null;
      }, "ListStatus", false, "request=%s", request);
    } catch (Exception e) {
      resultStream.fail(e);
    } finally {
      resultStream.complete();
    }
  }

  @Override
  public void listStatusPartial(ListStatusPartialPRequest request,
                                StreamObserver<ListStatusPartialPResponse> responseObserver) {
    ListStatusContext context = ListStatusContext.create(request.getOptions().toBuilder());
    ListStatusPartialResultStream resultStream =
        new ListStatusPartialResultStream(responseObserver, context);
    try {
      RpcUtils.callAndReturn(LOG, () -> {
        AlluxioURI pathUri = getAlluxioURI(request.getPath());
        mFileSystemMaster.listStatus(pathUri,
            context.withTracker(new GrpcCallTracker(responseObserver)),
            resultStream);
        return null;
      }, "ListStatus", false, "request=%s", request);
    } catch (Exception e) {
      resultStream.onError(e);
    } finally {
      resultStream.complete();
    }
  }

  @Override
  public void mount(MountPRequest request, StreamObserver<MountPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      MountContext mountContext = MountContext.create(request.getOptions().toBuilder())
          .withTracker(new GrpcCallTracker(responseObserver));
      // the mount execution process is recorded so that
      // when an exception occurs during mounting, the user can get detailed debugging messages
      try {
        mFileSystemMaster.mount(new AlluxioURI(request.getAlluxioPath()),
            new AlluxioURI(request.getUfsPath()), mountContext);
      } catch (Exception e) {
        Recorder recorder = mountContext.getRecorder();
        recorder.record(e.getMessage());
        // put the messages in an exception and let it carry over to the user
        throw new AlluxioException(String.join("\n", recorder.takeRecords()), e);
      }
      return MountPResponse.newBuilder().build();
    }, "Mount", "request=%s", responseObserver, request);
  }

  @Override
  public void updateMount(UpdateMountPRequest request,
      StreamObserver<UpdateMountPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      mFileSystemMaster.updateMount(new AlluxioURI(request.getAlluxioPath()),
          MountContext.create(request.getOptions().toBuilder())
              .withTracker(new GrpcCallTracker(responseObserver)));
      return UpdateMountPResponse.newBuilder().build();
    }, "UpdateMount", "request=%s", responseObserver, request);
  }

  @Override
  public void getMountTable(GetMountTablePRequest request,
      StreamObserver<GetMountTablePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      // Set the checkUfs default to true to include ufs usage info, etc.,
      // which requires talking to UFS and comes at a cost.
      boolean checkUfs = request.hasCheckUfs() ? request.getCheckUfs() : true;
      Map<String, MountPointInfo> mountTableWire = mFileSystemMaster.getMountPointInfoSummary(
          checkUfs);
      Map<String, alluxio.grpc.MountPointInfo> mountTableProto = new HashMap<>();
      for (Map.Entry<String, MountPointInfo> entry : mountTableWire.entrySet()) {
        mountTableProto.put(entry.getKey(), GrpcUtils.toProto(entry.getValue()));
      }
      return GetMountTablePResponse.newBuilder().putAllMountPoints(mountTableProto).build();
    }, "GetMountTable", "request=%s", responseObserver, request);
  }

  @Override
  public void getSyncPathList(GetSyncPathListPRequest request,
      StreamObserver<GetSyncPathListPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      List<SyncPointInfo> pathList = mFileSystemMaster.getSyncPathList();
      List<alluxio.grpc.SyncPointInfo> syncPointInfoList =
          pathList.stream().map(SyncPointInfo::toProto).collect(Collectors.toList());
      return GetSyncPathListPResponse.newBuilder().addAllSyncPaths(syncPointInfoList).build();
    }, "getSyncPathList", "request=%s", responseObserver, request);
  }

  @Override
  public void remove(DeletePRequest request, StreamObserver<DeletePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      mFileSystemMaster.delete(pathUri, DeleteContext.create(request.getOptions().toBuilder())
          .withTracker(new GrpcCallTracker(responseObserver)));
      return DeletePResponse.newBuilder().build();
    }, "Remove", "request=%s", responseObserver, request);
  }

  @Override
  public void rename(RenamePRequest request, StreamObserver<RenamePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI srcPathUri = getAlluxioURI(request.getPath());
      AlluxioURI dstPathUri = getAlluxioURI(request.getDstPath());
      mFileSystemMaster.rename(srcPathUri, dstPathUri,
          RenameContext.create(request.getOptions().toBuilder())
              .withTracker(new GrpcCallTracker(responseObserver)));
      return RenamePResponse.newBuilder().build();
    }, "Rename", "request=%s", responseObserver, request);
  }

  @Override
  public void reverseResolve(ReverseResolvePRequest request,
      StreamObserver<ReverseResolvePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI ufsUri = new AlluxioURI(request.getUfsUri());
      AlluxioURI alluxioPath = mFileSystemMaster.reverseResolve(ufsUri);
      return ReverseResolvePResponse.newBuilder().setAlluxioPath(alluxioPath.getPath()).build();
    }, "ReverseResolve", "request=%s", responseObserver, request);
  }

  @Override
  public void scheduleAsyncPersistence(ScheduleAsyncPersistencePRequest request,
      StreamObserver<ScheduleAsyncPersistencePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      mFileSystemMaster.scheduleAsyncPersistence(new AlluxioURI(request.getPath()),
          ScheduleAsyncPersistenceContext.create(request.getOptions().toBuilder()));
      return ScheduleAsyncPersistencePResponse.newBuilder().build();
    }, "ScheduleAsyncPersist", "request=%s", responseObserver, request);
  }

  @Override
  public void setAttribute(SetAttributePRequest request,
      StreamObserver<SetAttributePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      mFileSystemMaster.setAttribute(pathUri,
          SetAttributeContext.create(request.getOptions().toBuilder())
              .withTracker(new GrpcCallTracker(responseObserver)));
      return SetAttributePResponse.newBuilder().build();
    }, "SetAttribute", "request=%s", responseObserver, request);
  }

  @Override
  public void startSync(StartSyncPRequest request,
      StreamObserver<StartSyncPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      mFileSystemMaster.startSync(new AlluxioURI(request.getPath()));
      return StartSyncPResponse.newBuilder().build();
    }, "startSync", "request=%s", responseObserver, request);
  }

  @Override
  public void stopSync(StopSyncPRequest request,
      StreamObserver<StopSyncPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      mFileSystemMaster.stopSync(new AlluxioURI(request.getPath()));
      return StopSyncPResponse.newBuilder().build();
    }, "stopSync", "request=%s", responseObserver, request);
  }

  @Override
  public void unmount(UnmountPRequest request, StreamObserver<UnmountPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      mFileSystemMaster.unmount(new AlluxioURI(request.getAlluxioPath()));
      return UnmountPResponse.newBuilder().build();
    }, "Unmount", "request=%s", responseObserver, request);
  }

  @Override
  public void updateUfsMode(UpdateUfsModePRequest request,
      StreamObserver<UpdateUfsModePResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      UfsMode ufsMode;
      switch (request.getOptions().getUfsMode()) {
        case NO_ACCESS:
          ufsMode = UfsMode.NO_ACCESS;
          break;
        case READ_ONLY:
          ufsMode = UfsMode.READ_ONLY;
          break;
        default:
          ufsMode = UfsMode.READ_WRITE;
          break;
      }
      mFileSystemMaster.updateUfsMode(new AlluxioURI(request.getUfsPath()), ufsMode);
      return UpdateUfsModePResponse.newBuilder().build();
    }, "UpdateUfsMode", "request=%s", responseObserver, request);
  }

  @Override
  public void setAcl(SetAclPRequest request, StreamObserver<SetAclPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      AlluxioURI pathUri = getAlluxioURI(request.getPath());
      mFileSystemMaster.setAcl(pathUri, request.getAction(),
          request.getEntriesList().stream().map(GrpcUtils::fromProto).collect(Collectors.toList()),
          SetAclContext.create(request.getOptions().toBuilder())
              .withTracker(new GrpcCallTracker(responseObserver)));
      return SetAclPResponse.newBuilder().build();
    }, "setAcl", "request=%s", responseObserver, request);
  }

  @Override
  public void getStateLockHolders(GetStateLockHoldersPRequest request,
      StreamObserver<GetStateLockHoldersPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      final Collection<String> holders = mFileSystemMaster.getStateLockSharedWaitersAndHolders();
      return GetStateLockHoldersPResponse.newBuilder().addAllThreads(holders).build();
    }, "getStateLockHolders", "request=%s", responseObserver, request);
  }

  @Override
  public void needsSync(NeedsSyncRequest request,
                        StreamObserver<NeedsSyncResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      mFileSystemMaster.needsSync(new AlluxioURI(request.getPath()));
      return NeedsSyncResponse.getDefaultInstance();
    }, "NeedsSync", true, "request=%s", responseObserver, request);
  }

  @Override
  public void submitJob(SubmitJobPRequest request,
      StreamObserver<SubmitJobPResponse> responseObserver) {

    RpcUtils.call(LOG, () -> {
      JobRequest jobRequest;
      try {
        jobRequest = (JobRequest) SerializationUtils.deserialize(request
            .getRequestBody()
            .toByteArray());
      } catch (Exception e) {
        throw new IllegalArgumentException("fail to parse job request", e);
      }
      Job<?> job = JobFactoryProducer.create(jobRequest, mFileSystemMaster).create();
      boolean submitted = mScheduler.submitJob(job);
      SubmitJobPResponse.Builder builder = SubmitJobPResponse.newBuilder();
      if (submitted) {
        builder.setJobId(job.getJobId());
      }
      return builder.build();
    }, "submitJob", "request=%s", responseObserver, request);
  }

  @Override
  public void stopJob(StopJobPRequest request,
      StreamObserver<StopJobPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      boolean stopped = mScheduler.stopJob(JobDescription.from(request.getJobDescription()));
      return alluxio.grpc.StopJobPResponse.newBuilder()
          .setJobStopped(stopped)
          .build();
    }, "stopJob", "request=%s", responseObserver, request);
  }

  @Override
  public void getJobProgress(GetJobProgressPRequest request,
      StreamObserver<GetJobProgressPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      JobProgressReportFormat format = JobProgressReportFormat.TEXT;
      if (request.hasOptions() && request.getOptions().hasFormat()) {
        format = request.getOptions().getFormat();
      }
      boolean verbose = false;
      if (request.hasOptions() && request.getOptions().hasVerbose()) {
        verbose = request.getOptions().getVerbose();
      }
      return GetJobProgressPResponse.newBuilder()
          .setProgressReport(mScheduler.getJobProgress(
              JobDescription.from(request.getJobDescription()), format, verbose))
          .build();
    }, "getJobProgress", "request=%s", responseObserver, request);
  }

  /**
   * Helper to return {@link AlluxioURI} from transport URI.
   *
   * @param uriStr transport uri string
   * @return a {@link AlluxioURI} instance
   */
  private AlluxioURI getAlluxioURI(String uriStr) {
    return new AlluxioURI(uriStr);
  }

  @Override
  public void syncMetadata(
      SyncMetadataPRequest request,
      StreamObserver<SyncMetadataPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      return mFileSystemMaster.syncMetadata(
          new AlluxioURI(request.getPath()),
          SyncMetadataContext.create(request.getOptions().toBuilder()));
    }, "syncMetadata", "request=%s", responseObserver, request);
  }

  @Override
  public void syncMetadataAsync(
      SyncMetadataPRequest request,
      StreamObserver<SyncMetadataAsyncPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      return mFileSystemMaster.syncMetadataAsync(
          new AlluxioURI(request.getPath()),
          SyncMetadataContext.create(request.getOptions().toBuilder()));
    }, "syncMetadataAsync", "request=%s", responseObserver, request);
  }

  @Override
  public void getSyncProgress(
      GetSyncProgressPRequest request,
      StreamObserver<GetSyncProgressPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      return mFileSystemMaster.getSyncProgress(
          request.getTaskGroupId());
    }, "syncMetadataAsync", "request=%s", responseObserver, request);
  }

  @Override
  public void cancelSyncMetadata(
      CancelSyncMetadataPRequest request,
      StreamObserver<CancelSyncMetadataPResponse> responseObserver) {
    RpcUtils.call(LOG, () -> {
      return mFileSystemMaster.cancelSyncMetadata(
          request.getTaskGroupId());
    }, "cancelSyncMetadata", "request=%s", responseObserver, request);
  }
}
