/**
 * 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.nio.ByteBuffer;
import static org.junit.Assert.*;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.tez.common.JavaOptsChecker;
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.Vertex.VertexExecutionContext;
import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
import org.apache.tez.dag.api.records.DAGProtos.EdgePlan;
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.PlanVertexType;
import org.apache.tez.dag.api.records.DAGProtos.VertexExecutionContextProto;
import org.apache.tez.dag.api.records.DAGProtos.VertexPlan;
import org.apache.tez.serviceplugins.api.ContainerLauncherDescriptor;
import org.apache.tez.serviceplugins.api.ServicePluginsDescriptor;
import org.apache.tez.serviceplugins.api.TaskCommunicatorDescriptor;
import org.apache.tez.serviceplugins.api.TaskSchedulerDescriptor;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

// based on TestDAGLocationHint
public class TestDAGPlan {
  @Rule
  public TemporaryFolder tempFolder = new TemporaryFolder(); //TODO: doesn't seem to be deleting this folder automatically as expected.

  @Test(timeout = 5000)
  public void testBasicJobPlanSerde() throws IOException {

    DAGPlan job = DAGPlan.newBuilder()
       .setName("test")
       .addVertex(
           VertexPlan.newBuilder()
             .setName("vertex1")
             .setType(PlanVertexType.NORMAL)
             .addTaskLocationHint(PlanTaskLocationHint.newBuilder().addHost("machineName").addRack("rack1").build())
             .setTaskConfig(
                 PlanTaskConfiguration.newBuilder()
                   .setNumTasks(2)
                   .setVirtualCores(4)
                   .setMemoryMb(1024)
                   .setJavaOpts("")
                   .setTaskModule("x.y")
                   .build())
             .build())
        .build();
   File file = tempFolder.newFile("jobPlan");
   FileOutputStream outStream = null;
   try {
     outStream = new FileOutputStream(file);
     job.writeTo(outStream);
   }
   finally {
     if(outStream != null){
       outStream.close();
     }
   }

   DAGPlan inJob;
   FileInputStream inputStream;
   try {
     inputStream = new FileInputStream(file);
     inJob = DAGPlan.newBuilder().mergeFrom(inputStream).build();
   }
   finally {
     outStream.close();
   }

   Assert.assertEquals(job, inJob);
  }

  @Test(timeout = 5000)
  public void testEdgeManagerSerde() {
    DAG dag = DAG.create("DAG-testEdgeManagerSerde");
    ProcessorDescriptor pd1 = ProcessorDescriptor.create("processor1")
        .setUserPayload(UserPayload.create(ByteBuffer.wrap("processor1Bytes".getBytes())));
    ProcessorDescriptor pd2 = ProcessorDescriptor.create("processor2")
        .setUserPayload(UserPayload.create(ByteBuffer.wrap("processor2Bytes".getBytes())));
    Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
    Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
    v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());
    v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());

    InputDescriptor inputDescriptor = InputDescriptor.create("input")
        .setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
    OutputDescriptor outputDescriptor = OutputDescriptor.create("output")
        .setUserPayload(UserPayload.create(ByteBuffer.wrap("outputBytes".getBytes())));
    Edge edge = Edge.create(v1, v2, EdgeProperty.create(
        EdgeManagerPluginDescriptor.create("emClass").setUserPayload(
            UserPayload.create(ByteBuffer.wrap("emPayload".getBytes()))),
        DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL, outputDescriptor, inputDescriptor));

    dag.addVertex(v1).addVertex(v2).addEdge(edge);

    DAGPlan dagProto = dag.createDag(new TezConfiguration(), null, null, null, true);

    EdgeProperty edgeProperty = DagTypeConverters.createEdgePropertyMapFromDAGPlan(dagProto
        .getEdgeList().get(0));

    EdgeManagerPluginDescriptor emDesc = edgeProperty.getEdgeManagerDescriptor();
    Assert.assertNotNull(emDesc);
    Assert.assertEquals("emClass", emDesc.getClassName());
    Assert.assertTrue(
        Arrays.equals("emPayload".getBytes(), emDesc.getUserPayload().deepCopyAsArray()));
  }

  @Test(timeout = 5000)
  public void testUserPayloadSerde() {
    DAG dag = DAG.create("DAG-testUserPayloadSerde");
    ProcessorDescriptor pd1 = ProcessorDescriptor.create("processor1").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor1Bytes".getBytes())));
    ProcessorDescriptor pd2 = ProcessorDescriptor.create("processor2").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor2Bytes".getBytes())));
    Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
    Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
    v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());
    v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());

    InputDescriptor inputDescriptor = InputDescriptor.create("input").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
    OutputDescriptor outputDescriptor = OutputDescriptor.create("output").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("outputBytes".getBytes())));
    Edge edge = Edge.create(v1, v2, EdgeProperty.create(
        DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
        SchedulingType.SEQUENTIAL, outputDescriptor, inputDescriptor));

    dag.addVertex(v1).addVertex(v2).addEdge(edge);

    DAGPlan dagProto = dag.createDag(new TezConfiguration(), null, null, null, true);

    assertEquals(2, dagProto.getVertexCount());
    assertEquals(1, dagProto.getEdgeCount());

    VertexPlan v1Proto = dagProto.getVertex(0);
    VertexPlan v2Proto = dagProto.getVertex(1);
    EdgePlan edgeProto = dagProto.getEdge(0);

    assertEquals("processor1Bytes", new String(v1Proto.getProcessorDescriptor().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("processor1", v1Proto.getProcessorDescriptor().getClassName());

    assertEquals("processor2Bytes", new String(v2Proto.getProcessorDescriptor().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("processor2", v2Proto.getProcessorDescriptor().getClassName());

    assertEquals("inputBytes", new String(edgeProto.getEdgeDestination().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("input", edgeProto.getEdgeDestination().getClassName());

    assertEquals("outputBytes", new String(edgeProto.getEdgeSource().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("output", edgeProto.getEdgeSource().getClassName());

    EdgeProperty edgeProperty = DagTypeConverters
        .createEdgePropertyMapFromDAGPlan(dagProto.getEdgeList().get(0));

    byte[] ib = edgeProperty.getEdgeDestination().getUserPayload().deepCopyAsArray();
    assertEquals("inputBytes", new String(ib));
    assertEquals("input", edgeProperty.getEdgeDestination().getClassName());

    byte[] ob = edgeProperty.getEdgeSource().getUserPayload().deepCopyAsArray();
    assertEquals("outputBytes", new String(ob));
    assertEquals("output", edgeProperty.getEdgeSource().getClassName());
  }

  @Test(timeout = 5000)
  public void userVertexOrderingIsMaintained() {
    DAG dag = DAG.create("DAG-userVertexOrderingIsMaintained");
    ProcessorDescriptor pd1 = ProcessorDescriptor.create("processor1").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor1Bytes".getBytes())));
    ProcessorDescriptor pd2 = ProcessorDescriptor.create("processor2").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor2Bytes".getBytes())));
    ProcessorDescriptor pd3 = ProcessorDescriptor.create("processor3").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor3Bytes".getBytes())));
    Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
    Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
    Vertex v3 = Vertex.create("v3", pd3, 1, Resource.newInstance(1024, 1));
    v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());
    v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());
    v3.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());

    InputDescriptor inputDescriptor = InputDescriptor.create("input").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
    OutputDescriptor outputDescriptor = OutputDescriptor.create("output").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("outputBytes".getBytes())));
    Edge edge = Edge.create(v1, v2, EdgeProperty.create(
        DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
        SchedulingType.SEQUENTIAL, outputDescriptor, inputDescriptor));

    dag.addVertex(v1).addVertex(v2).addEdge(edge).addVertex(v3);

    DAGPlan dagProto = dag.createDag(new TezConfiguration(), null, null, null, true);

    assertEquals(3, dagProto.getVertexCount());
    assertEquals(1, dagProto.getEdgeCount());

    VertexPlan v1Proto = dagProto.getVertex(0);
    VertexPlan v2Proto = dagProto.getVertex(1);
    VertexPlan v3Proto = dagProto.getVertex(2);
    EdgePlan edgeProto = dagProto.getEdge(0);

    // either v1 or v2 will be on top based on topological order 
    String v1ProtoPayload = new String(v1Proto.getProcessorDescriptor().getTezUserPayload().getUserPayload().toByteArray());
    String v2ProtoPayload = new String(v2Proto.getProcessorDescriptor().getTezUserPayload().getUserPayload().toByteArray());
    assertTrue(v1ProtoPayload.equals("processor1Bytes") || v1ProtoPayload.equals("processor3Bytes"));
    assertTrue(v2ProtoPayload.equals("processor1Bytes") || v2ProtoPayload.equals("processor3Bytes"));
    assertTrue(v1Proto.getProcessorDescriptor().getClassName().equals("processor1") ||
        v1Proto.getProcessorDescriptor().getClassName().equals("processor3"));
    assertTrue(v2Proto.getProcessorDescriptor().getClassName().equals("processor1") ||
        v2Proto.getProcessorDescriptor().getClassName().equals("processor3"));

    assertEquals("processor2Bytes", new String(v3Proto.getProcessorDescriptor().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("processor2", v3Proto.getProcessorDescriptor().getClassName());

    assertEquals("inputBytes", new String(edgeProto.getEdgeDestination().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("input", edgeProto.getEdgeDestination().getClassName());

    assertEquals("outputBytes", new String(edgeProto.getEdgeSource().getTezUserPayload()
        .getUserPayload().toByteArray()));
    assertEquals("output", edgeProto.getEdgeSource().getClassName());

    EdgeProperty edgeProperty = DagTypeConverters
        .createEdgePropertyMapFromDAGPlan(dagProto.getEdgeList().get(0));

    byte[] ib = edgeProperty.getEdgeDestination().getUserPayload().deepCopyAsArray();
    assertEquals("inputBytes", new String(ib));
    assertEquals("input", edgeProperty.getEdgeDestination().getClassName());

    byte[] ob = edgeProperty.getEdgeSource().getUserPayload().deepCopyAsArray();
    assertEquals("outputBytes", new String(ob));
    assertEquals("output", edgeProperty.getEdgeSource().getClassName());
  }

  @Test (timeout=5000)
  public void testCredentialsSerde() {
    DAG dag = DAG.create("DAG-testCredentialsSerde");
    ProcessorDescriptor pd1 = ProcessorDescriptor.create("processor1").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor1Bytes".getBytes())));
    ProcessorDescriptor pd2 = ProcessorDescriptor.create("processor2").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor2Bytes".getBytes())));
    Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
    Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
    v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());
    v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());

    InputDescriptor inputDescriptor = InputDescriptor.create("input").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
    OutputDescriptor outputDescriptor = OutputDescriptor.create("output").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("outputBytes".getBytes())));
    Edge edge = Edge.create(v1, v2, EdgeProperty.create(
        DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
        SchedulingType.SEQUENTIAL, outputDescriptor, inputDescriptor));

    dag.addVertex(v1).addVertex(v2).addEdge(edge);

    Credentials dagCredentials = new Credentials();
    Token<TokenIdentifier> token1 = new Token<TokenIdentifier>();
    Token<TokenIdentifier> token2 = new Token<TokenIdentifier>();
    dagCredentials.addToken(new Text("Token1"), token1);
    dagCredentials.addToken(new Text("Token2"), token2);
    
    dag.setCredentials(dagCredentials);

    DAGPlan dagProto = dag.createDag(new TezConfiguration(), null, null, null, true);

    assertTrue(dagProto.hasCredentialsBinary());
    
    Credentials fetchedCredentials = DagTypeConverters.convertByteStringToCredentials(dagProto
        .getCredentialsBinary());
    
    assertEquals(2, fetchedCredentials.numberOfTokens());
    assertNotNull(fetchedCredentials.getToken(new Text("Token1")));
    assertNotNull(fetchedCredentials.getToken(new Text("Token2")));
  }

  @Test(timeout = 5000)
  public void testInvalidExecContext_1() {
    DAG dag = DAG.create("DAG-testInvalidExecContext_1");
    dag.setExecutionContext(VertexExecutionContext.createExecuteInAm(true));
    Vertex v1 = Vertex.create("testvertex", ProcessorDescriptor.create("processor1"), 1);
    dag.addVertex(v1);

    try {
      dag.createDag(new TezConfiguration(false), null, null, null, true);
      fail("Expecting dag create to fail due to invalid ServicePluginDescriptor");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains("AM execution"));
    }

    dag.setExecutionContext(VertexExecutionContext.createExecuteInContainers(true));

    try {
      dag.createDag(new TezConfiguration(false), null, null, null, true);
      fail("Expecting dag create to fail due to invalid ServicePluginDescriptor");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains("container execution"));
    }

  }

  @Test(timeout = 5000)
  public void testInvalidExecContext_2() {

    ServicePluginsDescriptor servicePluginsDescriptor = ServicePluginsDescriptor
        .create(false,
            new TaskSchedulerDescriptor[]{TaskSchedulerDescriptor.create("plugin", null)},
            new ContainerLauncherDescriptor[]{ContainerLauncherDescriptor.create("plugin", null)},
            new TaskCommunicatorDescriptor[]{TaskCommunicatorDescriptor.create("plugin", null)});

    VertexExecutionContext validExecContext = VertexExecutionContext.create("plugin", "plugin",
        "plugin");
    VertexExecutionContext invalidExecContext1 =
        VertexExecutionContext.create("invalidplugin", "plugin", "plugin");
    VertexExecutionContext invalidExecContext2 =
        VertexExecutionContext.create("plugin", "invalidplugin", "plugin");
    VertexExecutionContext invalidExecContext3 =
        VertexExecutionContext.create("plugin", "plugin", "invalidplugin");


    DAG dag = DAG.create("DAG-testInvalidExecContext_2");
    dag.setExecutionContext(VertexExecutionContext.createExecuteInContainers(true));
    Vertex v1 = Vertex.create("testvertex", ProcessorDescriptor.create("processor1"), 1);
    dag.addVertex(v1);

    // Should succeed. Default context is containers.
    dag.createDag(new TezConfiguration(false), null, null, null, true,
        servicePluginsDescriptor, null);


    // Set execute in AM should fail
    v1.setExecutionContext(VertexExecutionContext.createExecuteInAm(true));
    try {
      dag.createDag(new TezConfiguration(false), null, null, null, true,
          servicePluginsDescriptor, null);
      fail("Expecting dag create to fail due to invalid ServicePluginDescriptor");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains("AM execution"));
    }

    // Valid context
    v1.setExecutionContext(validExecContext);
    dag.createDag(new TezConfiguration(false), null, null, null, true,
        servicePluginsDescriptor, null);

    // Invalid task scheduler
    v1.setExecutionContext(invalidExecContext1);
    try {
      dag.createDag(new TezConfiguration(false), null, null, null, true,
          servicePluginsDescriptor, null);
      fail("Expecting dag create to fail due to invalid ServicePluginDescriptor");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains("testvertex"));
      assertTrue(e.getMessage().contains("task scheduler"));
      assertTrue(e.getMessage().contains("invalidplugin"));
    }

    // Invalid ContainerLauncher
    v1.setExecutionContext(invalidExecContext2);
    try {
      dag.createDag(new TezConfiguration(false), null, null, null, true,
          servicePluginsDescriptor, null);
      fail("Expecting dag create to fail due to invalid ServicePluginDescriptor");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains("testvertex"));
      assertTrue(e.getMessage().contains("container launcher"));
      assertTrue(e.getMessage().contains("invalidplugin"));
    }

    // Invalid task comm
    v1.setExecutionContext(invalidExecContext3);
    try {
      dag.createDag(new TezConfiguration(false), null, null, null, true,
          servicePluginsDescriptor, null);
      fail("Expecting dag create to fail due to invalid ServicePluginDescriptor");
    } catch (IllegalStateException e) {
      assertTrue(e.getMessage().contains("testvertex"));
      assertTrue(e.getMessage().contains("task communicator"));
      assertTrue(e.getMessage().contains("invalidplugin"));
    }

  }

  @Test(timeout = 5000)
  public void testServiceDescriptorPropagation() {
    DAG dag = DAG.create("DAG-testServiceDescriptorPropagation");
    ProcessorDescriptor pd1 = ProcessorDescriptor.create("processor1").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor1Bytes".getBytes())));
    ProcessorDescriptor pd2 = ProcessorDescriptor.create("processor2").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("processor2Bytes".getBytes())));

    VertexExecutionContext defaultExecutionContext =
        VertexExecutionContext.create("plugin", "plugin", "plugin");
    VertexExecutionContext v1Context = VertexExecutionContext.createExecuteInAm(true);

    ServicePluginsDescriptor servicePluginsDescriptor = ServicePluginsDescriptor
        .create(true, new TaskSchedulerDescriptor[]{TaskSchedulerDescriptor.create("plugin", null)},
            new ContainerLauncherDescriptor[]{ContainerLauncherDescriptor.create("plugin", null)},
            new TaskCommunicatorDescriptor[]{TaskCommunicatorDescriptor.create("plugin", null)});

    Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1))
        .setExecutionContext(v1Context);
    Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
    v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());
    v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
        .addTaskLocalFiles(new HashMap<String, LocalResource>());

    InputDescriptor inputDescriptor = InputDescriptor.create("input").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
    OutputDescriptor outputDescriptor = OutputDescriptor.create("output").
        setUserPayload(UserPayload.create(ByteBuffer.wrap("outputBytes".getBytes())));
    Edge edge = Edge.create(v1, v2, EdgeProperty.create(
        DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
        SchedulingType.SEQUENTIAL, outputDescriptor, inputDescriptor));

    dag.addVertex(v1).addVertex(v2).addEdge(edge);
    dag.setExecutionContext(defaultExecutionContext);

    DAGPlan dagProto = dag.createDag(new TezConfiguration(), null, null, null, true,
        servicePluginsDescriptor, null);

    assertEquals(2, dagProto.getVertexCount());
    assertEquals(1, dagProto.getEdgeCount());

    assertTrue(dagProto.hasDefaultExecutionContext());
    VertexExecutionContextProto defaultContextProto = dagProto.getDefaultExecutionContext();
    assertFalse(defaultContextProto.getExecuteInContainers());
    assertFalse(defaultContextProto.getExecuteInAm());
    assertEquals("plugin", defaultContextProto.getTaskSchedulerName());
    assertEquals("plugin", defaultContextProto.getContainerLauncherName());
    assertEquals("plugin", defaultContextProto.getTaskCommName());

    VertexPlan v1Proto = dagProto.getVertex(0);
    assertTrue(v1Proto.hasExecutionContext());
    VertexExecutionContextProto v1ContextProto = v1Proto.getExecutionContext();
    assertFalse(v1ContextProto.getExecuteInContainers());
    assertTrue(v1ContextProto.getExecuteInAm());
    assertFalse(v1ContextProto.hasTaskSchedulerName());
    assertFalse(v1ContextProto.hasContainerLauncherName());
    assertFalse(v1ContextProto.hasTaskCommName());

    VertexPlan v2Proto = dagProto.getVertex(1);
    assertFalse(v2Proto.hasExecutionContext());
  }

  @Test(timeout = 5000)
  public void testInvalidJavaOpts() {
    DAG dag = DAG.create("DAG-testInvalidJavaOpts");
    ProcessorDescriptor pd1 = ProcessorDescriptor.create("processor1")
        .setUserPayload(UserPayload.create(ByteBuffer.wrap("processor1Bytes".getBytes())));
    Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
    v1.setTaskLaunchCmdOpts(" -XX:+UseG1GC ");

    dag.addVertex(v1);

    TezConfiguration conf = new TezConfiguration(false);
    conf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, "  -XX:+UseParallelGC ");
    try {
      dag.createDag(conf, null, null, null, true, null, new JavaOptsChecker());
      fail("Expected dag creation to fail for invalid java opts");
    } catch (TezUncheckedException e) {
      Assert.assertTrue(e.getMessage().contains("Invalid/conflicting GC options"));
    }

    // Should not fail as java opts valid
    conf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, "  -XX:-UseParallelGC ");
    dag.createDag(conf, null, null, null, true, null, new JavaOptsChecker());

    // Should not fail as no checker enabled
    conf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, "  -XX:+UseParallelGC ");
    dag.createDag(conf, null, null, null, true, null, null);

  }

}
