/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.tez.dag.api;

import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.Map.Entry;

import javax.annotation.Nullable;

import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DataInputByteBuffer;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.URL;
import org.apache.hadoop.yarn.api.records.impl.pb.LocalResourcePBImpl;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.tez.client.TezAppMasterStatus;
import org.apache.tez.common.TezCommonUtils;
import org.apache.tez.common.counters.CounterGroup;
import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.common.counters.TezCounters;
import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.client.StatusGetOpts;
import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolRPC.TezSessionStatusProto;
import org.apache.tez.dag.api.records.DAGProtos;
import org.apache.tez.dag.api.records.DAGProtos.ConfigurationProto;
import org.apache.tez.dag.api.records.DAGProtos.EdgePlan;
import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataMovementType;
import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataSourceType;
import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSchedulingType;
import org.apache.tez.dag.api.records.DAGProtos.PlanKeyValuePair;
import org.apache.tez.dag.api.records.DAGProtos.PlanLocalResource;
import org.apache.tez.dag.api.records.DAGProtos.PlanLocalResourcesProto;
import org.apache.tez.dag.api.records.DAGProtos.PlanLocalResourceType;
import org.apache.tez.dag.api.records.DAGProtos.PlanLocalResourceVisibility;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskConfiguration;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskLocationHint;
import org.apache.tez.dag.api.records.DAGProtos.RootInputLeafOutputProto;
import org.apache.tez.dag.api.records.DAGProtos.TezCounterGroupProto;
import org.apache.tez.dag.api.records.DAGProtos.TezCounterProto;
import org.apache.tez.dag.api.records.DAGProtos.TezCountersProto;
import org.apache.tez.dag.api.records.DAGProtos.TezEntityDescriptorProto;
import org.apache.tez.dag.api.records.DAGProtos.VertexLocationHintProto;

import com.google.common.collect.Lists;
import com.google.protobuf.ByteString;
import com.google.protobuf.ByteString.Output;

@Private
public class DagTypeConverters {

  public static PlanLocalResourceVisibility convertToDAGPlan(LocalResourceVisibility visibility){
    switch(visibility){
      case PUBLIC : return PlanLocalResourceVisibility.PUBLIC;
      case PRIVATE : return PlanLocalResourceVisibility.PRIVATE;
      case APPLICATION : return PlanLocalResourceVisibility.APPLICATION;
      default : throw new RuntimeException("unknown 'visibility': " + visibility);
    }
  }
  
  public static List<PlanLocalResource> convertToDAGPlan(Map<String, LocalResource> lrs) {
    List<PlanLocalResource> planLrs = Lists.newArrayListWithCapacity(lrs.size());
    for (Entry<String, LocalResource> entry : lrs.entrySet()) {
      PlanLocalResource.Builder localResourcesBuilder = PlanLocalResource.newBuilder();
      String key = entry.getKey();
      LocalResource lr = entry.getValue();
      localResourcesBuilder.setName(key);
      localResourcesBuilder.setUri(
        DagTypeConverters.convertToDAGPlan(lr.getResource()));
      localResourcesBuilder.setSize(lr.getSize());
      localResourcesBuilder.setTimeStamp(lr.getTimestamp());
      localResourcesBuilder.setType(
        DagTypeConverters.convertToDAGPlan(lr.getType()));
      localResourcesBuilder.setVisibility(
        DagTypeConverters.convertToDAGPlan(lr.getVisibility()));
      if (lr.getType() == LocalResourceType.PATTERN) {
        if (lr.getPattern() == null || lr.getPattern().isEmpty()) {
          throw new TezUncheckedException("LocalResource type set to pattern"
            + " but pattern is null or empty");
        }
        localResourcesBuilder.setPattern(lr.getPattern());
      }
      planLrs.add(localResourcesBuilder.build());
    }
    return planLrs;
  }

  public static LocalResourceVisibility convertFromDAGPlan(PlanLocalResourceVisibility visibility){
    switch(visibility){
      case PUBLIC : return LocalResourceVisibility.PUBLIC;
      case PRIVATE : return LocalResourceVisibility.PRIVATE;
      case APPLICATION : return LocalResourceVisibility.APPLICATION;
      default : throw new RuntimeException("unknown 'visibility': " + visibility);
    }
  }

  public static PlanEdgeDataSourceType convertToDAGPlan(DataSourceType sourceType){
    switch(sourceType){
      case PERSISTED : return PlanEdgeDataSourceType.PERSISTED;
      case PERSISTED_RELIABLE : return PlanEdgeDataSourceType.PERSISTED_RELIABLE;
      case EPHEMERAL :  return PlanEdgeDataSourceType.EPHEMERAL;
      default : throw new RuntimeException("unknown 'dataSourceType': " + sourceType);
    }
  }

  public static DataSourceType convertFromDAGPlan(PlanEdgeDataSourceType sourceType){
    switch(sourceType){
      case PERSISTED : return DataSourceType.PERSISTED;
      case PERSISTED_RELIABLE : return DataSourceType.PERSISTED_RELIABLE;
      case EPHEMERAL :  return DataSourceType.EPHEMERAL;
      default : throw new RuntimeException("unknown 'dataSourceType': " + sourceType);
    }
  }

  public static PlanEdgeDataMovementType convertToDAGPlan(DataMovementType type){
    switch(type){
      case ONE_TO_ONE : return PlanEdgeDataMovementType.ONE_TO_ONE;
      case BROADCAST : return PlanEdgeDataMovementType.BROADCAST;
      case SCATTER_GATHER : return PlanEdgeDataMovementType.SCATTER_GATHER;
      case CUSTOM: return PlanEdgeDataMovementType.CUSTOM;
      default : throw new RuntimeException("unknown 'dataMovementType': " + type);
    }
  }

  public static DataMovementType convertFromDAGPlan(PlanEdgeDataMovementType type){
    switch(type){
      case ONE_TO_ONE : return DataMovementType.ONE_TO_ONE;
      case BROADCAST : return DataMovementType.BROADCAST;
      case SCATTER_GATHER : return DataMovementType.SCATTER_GATHER;
      default : throw new IllegalArgumentException("unknown 'dataMovementType': " + type);
    }
  }

  public static PlanEdgeSchedulingType convertToDAGPlan(SchedulingType type){
    switch(type){
      case SEQUENTIAL : return PlanEdgeSchedulingType.SEQUENTIAL;
      case CONCURRENT : return PlanEdgeSchedulingType.CONCURRENT;
      default : throw new RuntimeException("unknown 'SchedulingType': " + type);
    }
  }

  public static SchedulingType convertFromDAGPlan(PlanEdgeSchedulingType type){
    switch(type){
      case SEQUENTIAL : return SchedulingType.SEQUENTIAL;
      case CONCURRENT : return SchedulingType.CONCURRENT;
      default : throw new IllegalArgumentException("unknown 'SchedulingType': " + type);
    }
  }

  public static PlanLocalResourceType convertToDAGPlan(LocalResourceType type) {
    switch(type){
    case ARCHIVE : return PlanLocalResourceType.ARCHIVE;
    case FILE : return PlanLocalResourceType.FILE;
    case PATTERN : return PlanLocalResourceType.PATTERN;
    default : throw new IllegalArgumentException("unknown 'type': " + type);
    }
  }

  public static LocalResourceType convertFromDAGPlan(PlanLocalResourceType type) {
    switch(type){
    case ARCHIVE : return LocalResourceType.ARCHIVE;
    case FILE : return LocalResourceType.FILE;
    case PATTERN : return LocalResourceType.PATTERN;
    default : throw new IllegalArgumentException("unknown 'type': " + type);
    }
  }

  public static VertexLocationHint convertFromDAGPlan(
      List<PlanTaskLocationHint> locationHints) {

    List<TaskLocationHint> outputList = new ArrayList<TaskLocationHint>();

    for(PlanTaskLocationHint inputHint : locationHints){
      TaskLocationHint outputHint = TaskLocationHint.createTaskLocationHint(
          new HashSet<String>(inputHint.getHostList()),
          new HashSet<String>(inputHint.getRackList()));
      outputList.add(outputHint);
    }
    return VertexLocationHint.create(outputList);
  }
  
  // notes re HDFS URL handling:
  //   Resource URLs in the protobuf message are strings of the form hdfs://host:port/path
  //   org.apache.hadoop.fs.Path.Path  is actually a URI type that allows any scheme
  //   org.apache.hadoop.yarn.api.records.URL is a URL type used by YARN.
  //   java.net.URL cannot be used out of the box as it rejects unknown schemes such as HDFS.

  public static String convertToDAGPlan(URL resource) {
    // see above notes on HDFS URL handling
    return resource.getScheme() + "://" + resource.getHost()
        + ":" + resource.getPort() + resource.getFile();
  }

  public static Map<String, LocalResource> createLocalResourceMapFromDAGPlan(
      List<PlanLocalResource> localResourcesList) {
    Map<String, LocalResource> map = new HashMap<String, LocalResource>();
    for(PlanLocalResource res : localResourcesList){
      LocalResource r = new LocalResourcePBImpl();

      //NOTE: have to check every optional field in protobuf generated classes for existence before accessing
      //else we will receive a default value back, eg ""
      if(res.hasPattern()){
        r.setPattern(res.getPattern());
      }
      r.setResource(ConverterUtils.getYarnUrlFromPath(new Path(res.getUri())));  // see above notes on HDFS URL handling
      r.setSize(res.getSize());
      r.setTimestamp(res.getTimeStamp());
      r.setType(DagTypeConverters.convertFromDAGPlan(res.getType()));
      r.setVisibility(DagTypeConverters.convertFromDAGPlan(res.getVisibility()));
      map.put(res.getName(), r);
    }
    return map;
  }

  public static Map<String, String> createEnvironmentMapFromDAGPlan(
      List<PlanKeyValuePair> environmentSettingList) {

    Map<String, String> map = new HashMap<String, String>();
    for(PlanKeyValuePair setting : environmentSettingList){
      map.put(setting.getKey(), setting.getValue());
    }

    return map;
  }

  public static Map<String, EdgePlan> createEdgePlanMapFromDAGPlan(List<EdgePlan> edgeList){
    Map<String, EdgePlan> edgePlanMap =
        new HashMap<String, EdgePlan>();
    for(EdgePlan edgePlanItem : edgeList){
      edgePlanMap.put(edgePlanItem.getId(), edgePlanItem);
    }
    return edgePlanMap;
  }

  public static EdgeProperty createEdgePropertyMapFromDAGPlan(EdgePlan edge) {
    if (edge.getDataMovementType() == PlanEdgeDataMovementType.CUSTOM) {
      return EdgeProperty.create(
          (edge.hasEdgeManager() ?
              convertEdgeManagerPluginDescriptorFromDAGPlan(edge.getEdgeManager()) : null),
          convertFromDAGPlan(edge.getDataSourceType()),
          convertFromDAGPlan(edge.getSchedulingType()),
          convertOutputDescriptorFromDAGPlan(edge.getEdgeSource()),
          convertInputDescriptorFromDAGPlan(edge.getEdgeDestination())
      );
    } else {
      return EdgeProperty.create(
          convertFromDAGPlan(edge.getDataMovementType()),
          convertFromDAGPlan(edge.getDataSourceType()),
          convertFromDAGPlan(edge.getSchedulingType()),
          convertOutputDescriptorFromDAGPlan(edge.getEdgeSource()),
          convertInputDescriptorFromDAGPlan(edge.getEdgeDestination())
      );
    }
  }

  public static Resource createResourceRequestFromTaskConfig(
      PlanTaskConfiguration taskConfig) {
    return Resource.newInstance(taskConfig.getMemoryMb(), taskConfig.getVirtualCores());
  }

  public static Map<String, String> convertConfFromProto(
      ConfigurationProto confProto) {
    List<PlanKeyValuePair> settingList = confProto.getConfKeyValuesList();
    Map<String, String> map = new HashMap<String, String>();
    for(PlanKeyValuePair setting: settingList){
      map.put(setting.getKey(), setting.getValue());
    }
    return map;
  }

  public static TezEntityDescriptorProto convertToDAGPlan(
      EntityDescriptor<?> descriptor) {
    TezEntityDescriptorProto.Builder builder = TezEntityDescriptorProto
        .newBuilder();
    builder.setClassName(descriptor.getClassName());

    UserPayload userPayload = descriptor.getUserPayload();
    if (userPayload != null && userPayload.hasPayload()) {
      builder.setUserPayload(ByteString.copyFrom(descriptor.getUserPayload().getPayload()));
      builder.setVersion(userPayload.getVersion());
    }
    if (descriptor.getHistoryText() != null) {
      try {
        builder.setHistoryText(TezCommonUtils.compressByteArrayToByteString(
            descriptor.getHistoryText().getBytes()));
      } catch (IOException e) {
        throw new TezUncheckedException(e);
      }
    }
    return builder.build();
  }

  public static String getHistoryTextFromProto(TezEntityDescriptorProto proto) {
    if (!proto.hasHistoryText()) {
      return null;
    }
    try {
      return new String(TezCommonUtils.decompressByteStringToByteArray(proto.getHistoryText()));
    } catch (IOException e) {
      throw new TezUncheckedException(e);
    }
  }


  public static RootInputLeafOutputProto convertToDAGPlan(
      RootInputLeafOutput<? extends EntityDescriptor<?>, ? extends EntityDescriptor<?>> rootIO) {
    RootInputLeafOutputProto.Builder builder = RootInputLeafOutputProto.newBuilder();
    builder.setName(rootIO.getName());
    builder.setIODescriptor(convertToDAGPlan(rootIO.getIODescriptor()));
    if (rootIO.getControllerDescriptor() != null) {
      builder.setControllerDescriptor(convertToDAGPlan(rootIO.getControllerDescriptor()));
    }
    return builder.build();
  }

  private static UserPayload convertTezUserPayloadFromDAGPlan(
      TezEntityDescriptorProto proto) {
    UserPayload userPayload = null;
    if (proto.hasUserPayload()) {
      userPayload =
          UserPayload.create(proto.getUserPayload().asReadOnlyByteBuffer(), proto.getVersion());
    } else {
      userPayload = UserPayload.create(null, -1);
    }
    return userPayload;
  }

  public static InputDescriptor convertInputDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return InputDescriptor.create(className).setUserPayload(payload);
  }

  public static OutputDescriptor convertOutputDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return OutputDescriptor.create(className).setUserPayload(payload);
  }

  public static InputInitializerDescriptor convertInputInitializerDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return InputInitializerDescriptor.create(className).setUserPayload(payload);
  }

  public static OutputCommitterDescriptor convertOutputCommitterDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return OutputCommitterDescriptor.create(className).setUserPayload(payload);
  }

  public static VertexManagerPluginDescriptor convertVertexManagerPluginDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return VertexManagerPluginDescriptor.create(className).setUserPayload(payload);
  }

  public static EdgeManagerPluginDescriptor convertEdgeManagerPluginDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return EdgeManagerPluginDescriptor.create(className).setUserPayload(payload);
  }

  public static ProcessorDescriptor convertProcessorDescriptorFromDAGPlan(
      TezEntityDescriptorProto proto) {
    String className = proto.getClassName();
    UserPayload payload = convertTezUserPayloadFromDAGPlan(proto);
    return ProcessorDescriptor.create(className).setUserPayload(payload);
  }

  public static TezAppMasterStatus convertTezSessionStatusFromProto(
      TezSessionStatusProto proto) {
    switch (proto) {
    case INITIALIZING:
      return TezAppMasterStatus.INITIALIZING;
    case READY:
      return TezAppMasterStatus.READY;
    case RUNNING:
      return TezAppMasterStatus.RUNNING;
    case SHUTDOWN:
      return TezAppMasterStatus.SHUTDOWN;
    }
    throw new TezUncheckedException("Could not convert to TezSessionStatus from"
        + " proto");
  }

  public static TezSessionStatusProto convertTezSessionStatusToProto(
      TezAppMasterStatus status) {
    switch (status) {
    case INITIALIZING:
      return TezSessionStatusProto.INITIALIZING;
    case READY:
      return TezSessionStatusProto.READY;
    case RUNNING:
      return TezSessionStatusProto.RUNNING;
    case SHUTDOWN:
      return TezSessionStatusProto.SHUTDOWN;
    }
    throw new TezUncheckedException("Could not convert TezSessionStatus to"
        + " proto");
  }


  public static PlanLocalResourcesProto convertFromLocalResources(
    Map<String, LocalResource> localResources) {
    PlanLocalResourcesProto.Builder builder =
      PlanLocalResourcesProto.newBuilder();
    for (Map.Entry<String, LocalResource> entry : localResources.entrySet()) {
      PlanLocalResource plr = convertLocalResourceToPlanLocalResource(
        entry.getKey(), entry.getValue());
      builder.addLocalResources(plr);
    }
    return builder.build();
  }

  public static Map<String, LocalResource> convertFromPlanLocalResources(
    PlanLocalResourcesProto proto) {
    Map<String, LocalResource> localResources =
      new HashMap<String, LocalResource>(proto.getLocalResourcesCount());
    for (PlanLocalResource plr : proto.getLocalResourcesList()) {
      String name = plr.getName();
      LocalResource lr = convertPlanLocalResourceToLocalResource(plr);
      localResources.put(name, lr);
    }
    return localResources;
  }

  public static PlanLocalResource convertLocalResourceToPlanLocalResource(
    String name, LocalResource lr) {
    PlanLocalResource.Builder localResourcesBuilder = PlanLocalResource.newBuilder();
    localResourcesBuilder.setName(name);
    localResourcesBuilder.setUri(
      DagTypeConverters.convertToDAGPlan(lr.getResource()));
    localResourcesBuilder.setSize(lr.getSize());
    localResourcesBuilder.setTimeStamp(lr.getTimestamp());
    localResourcesBuilder.setType(
      DagTypeConverters.convertToDAGPlan(lr.getType()));
    localResourcesBuilder.setVisibility(
      DagTypeConverters.convertToDAGPlan(lr.getVisibility()));
    if (lr.getType() == LocalResourceType.PATTERN) {
      if (lr.getPattern() == null || lr.getPattern().isEmpty()) {
        throw new TezUncheckedException("LocalResource type set to pattern"
          + " but pattern is null or empty");
      }
      localResourcesBuilder.setPattern(lr.getPattern());
    }
    return localResourcesBuilder.build();
  }

  public static LocalResource convertPlanLocalResourceToLocalResource(
      PlanLocalResource plr) {
    return LocalResource.newInstance(
      ConverterUtils.getYarnUrlFromPath(new Path(plr.getUri())),
      DagTypeConverters.convertFromDAGPlan(plr.getType()),
      DagTypeConverters.convertFromDAGPlan(plr.getVisibility()),
      plr.getSize(), plr.getTimeStamp(),
      plr.hasPattern() ? plr.getPattern() : null);
  }

  public static TezCounters convertTezCountersFromProto(TezCountersProto proto) {
    TezCounters counters = new TezCounters();
    for (TezCounterGroupProto counterGroupProto : proto.getCounterGroupsList()) {
      CounterGroup group = counters.addGroup(counterGroupProto.getName(),
        counterGroupProto.getDisplayName());
      for (TezCounterProto counterProto :
        counterGroupProto.getCountersList()) {
        TezCounter counter = group.findCounter(
          counterProto.getName(),
          counterProto.getDisplayName());
        counter.setValue(counterProto.getValue());
      }
    }
    return counters;
  }

  public static TezCountersProto convertTezCountersToProto(
      TezCounters counters) {
    TezCountersProto.Builder builder = TezCountersProto.newBuilder();
    Iterator<CounterGroup> groupIterator = counters.iterator();
    int groupIndex = 0;
    while (groupIterator.hasNext()) {
      CounterGroup counterGroup = groupIterator.next();
      TezCounterGroupProto.Builder groupBuilder =
        TezCounterGroupProto.newBuilder();
      groupBuilder.setName(counterGroup.getName());
      groupBuilder.setDisplayName(counterGroup.getDisplayName());
      Iterator<TezCounter> counterIterator = counterGroup.iterator();
      int counterIndex = 0;
      while (counterIterator.hasNext()) {
        TezCounter counter = counterIterator.next();
        TezCounterProto tezCounterProto = TezCounterProto.newBuilder()
          .setName(counter.getName())
          .setDisplayName(counter.getDisplayName())
          .setValue(counter.getValue())
          .build();
        groupBuilder.addCounters(counterIndex, tezCounterProto);
        ++counterIndex;
      }
      builder.addCounterGroups(groupIndex, groupBuilder.build());
      ++groupIndex;
    }
    return builder.build();
  }

  public static DAGProtos.StatusGetOptsProto convertStatusGetOptsToProto(
    StatusGetOpts statusGetOpts) {
    switch (statusGetOpts) {
      case GET_COUNTERS:
        return DAGProtos.StatusGetOptsProto.GET_COUNTERS;
    }
    throw new TezUncheckedException("Could not convert StatusGetOpts to"
      + " proto");
  }

  public static StatusGetOpts convertStatusGetOptsFromProto(
    DAGProtos.StatusGetOptsProto proto) {
    switch (proto) {
      case GET_COUNTERS:
        return StatusGetOpts.GET_COUNTERS;
    }
    throw new TezUncheckedException("Could not convert to StatusGetOpts from"
      + " proto");
  }

  public static List<DAGProtos.StatusGetOptsProto> convertStatusGetOptsToProto(
    Set<StatusGetOpts> statusGetOpts) {
    List<DAGProtos.StatusGetOptsProto> protos =
      new ArrayList<DAGProtos.StatusGetOptsProto>(statusGetOpts.size());
    for (StatusGetOpts opt : statusGetOpts) {
      protos.add(convertStatusGetOptsToProto(opt));
    }
    return protos;
  }

  public static Set<StatusGetOpts> convertStatusGetOptsFromProto(
      List<DAGProtos.StatusGetOptsProto> protoList) {
    Set<StatusGetOpts> opts = new TreeSet<StatusGetOpts>();
    for (DAGProtos.StatusGetOptsProto proto : protoList) {
      opts.add(convertStatusGetOptsFromProto(proto));
    }
    return opts;
  }

  public static ByteString convertCredentialsToProto(Credentials credentials) {
    if (credentials == null) {
      return null;
    }
    Output output = ByteString.newOutput();
    DataOutputStream dos = new DataOutputStream(output);
    try {
      credentials.writeTokenStorageToStream(dos);
      return output.toByteString();
    } catch (IOException e) {
      throw new TezUncheckedException("Failed to serialize Credentials", e);
    }
  }

  public static Credentials convertByteStringToCredentials(ByteString byteString) {
    if (byteString == null) {
      return null;
    }
    DataInputByteBuffer dib = new DataInputByteBuffer();
    dib.reset(byteString.asReadOnlyByteBuffer());
    Credentials credentials = new Credentials();
    try {
      credentials.readTokenStorageStream(dib);
      return credentials;
    } catch (IOException e) {
      throw new TezUncheckedException("Failed to deserialize Credentials", e);
    }
  }

  public static VertexLocationHint convertVertexLocationHintFromProto(
    VertexLocationHintProto proto) {
    List<TaskLocationHint> outputList = new ArrayList<TaskLocationHint>(
      proto.getTaskLocationHintsCount());
    for(PlanTaskLocationHint inputHint : proto.getTaskLocationHintsList()){
      TaskLocationHint outputHint = TaskLocationHint.createTaskLocationHint(
          new HashSet<String>(inputHint.getHostList()),
          new HashSet<String>(inputHint.getRackList()));
      outputList.add(outputHint);
    }

    return VertexLocationHint.create(outputList);
  }

  public static VertexLocationHintProto convertVertexLocationHintToProto(
      VertexLocationHint vertexLocationHint) {
    VertexLocationHintProto.Builder builder =
      VertexLocationHintProto.newBuilder();
    if (vertexLocationHint.getTaskLocationHints() != null) {
      for (TaskLocationHint taskLocationHint :
        vertexLocationHint.getTaskLocationHints()) {
        PlanTaskLocationHint.Builder taskLHBuilder =
          PlanTaskLocationHint.newBuilder();
        if (taskLocationHint.getHosts() != null) {
          taskLHBuilder.addAllHost(taskLocationHint.getHosts());
        }
        if (taskLocationHint.getRacks() != null) {
          taskLHBuilder.addAllRack(taskLocationHint.getRacks());
        }
        builder.addTaskLocationHints(taskLHBuilder.build());
      }
    }
    return builder.build();
  }

  public static UserPayload convertToTezUserPayload(@Nullable ByteBuffer payload, int version) {
    return UserPayload.create(payload, version);
  }

  @Nullable
  public static ByteBuffer convertFromTezUserPayload(@Nullable UserPayload payload) {
    if (payload == null) {
      return null;
    }
    return payload.getPayload();
  }

}