| /** |
| * 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.net.URI; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Deque; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.Stack; |
| |
| import org.apache.commons.collections4.BidiMap; |
| import org.apache.commons.collections4.bidimap.DualLinkedHashBidiMap; |
| import org.apache.hadoop.classification.InterfaceStability; |
| import org.apache.tez.client.CallerContext; |
| import org.apache.tez.common.JavaOptsChecker; |
| import org.apache.tez.dag.api.Vertex.VertexExecutionContext; |
| import org.apache.tez.dag.api.records.DAGProtos; |
| import org.apache.tez.serviceplugins.api.ServicePluginsDescriptor; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.hadoop.classification.InterfaceAudience.Private; |
| import org.apache.hadoop.classification.InterfaceAudience.Public; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.security.Credentials; |
| import org.apache.hadoop.yarn.api.records.LocalResource; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.tez.client.TezClientUtils; |
| import org.apache.tez.common.security.DAGAccessControls; |
| import org.apache.tez.common.TezCommonUtils; |
| import org.apache.tez.common.TezYARNUtils; |
| 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.VertexGroup.GroupInfo; |
| import org.apache.tez.dag.api.records.DAGProtos.ConfigurationProto; |
| 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.PlanGroupInputEdgeInfo; |
| import org.apache.tez.dag.api.records.DAGProtos.PlanKeyValuePair; |
| 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.PlanVertexGroupInfo; |
| import org.apache.tez.dag.api.records.DAGProtos.PlanVertexType; |
| import org.apache.tez.dag.api.records.DAGProtos.VertexPlan; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| |
| /** |
| * Top level entity that defines the DAG (Directed Acyclic Graph) representing |
| * the data flow graph. Consists of a set of Vertices and Edges connecting the |
| * vertices. Vertices represent transformations of data and edges represent |
| * movement of data between vertices. |
| */ |
| @Public |
| public class DAG { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(DAG.class); |
| |
| final BidiMap<String, Vertex> vertices = |
| new DualLinkedHashBidiMap<String, Vertex>(); |
| final Set<Edge> edges = Sets.newHashSet(); |
| final String name; |
| final Collection<URI> urisForCredentials = new HashSet<URI>(); |
| Credentials credentials = new Credentials(); |
| Set<VertexGroup> vertexGroups = Sets.newHashSet(); |
| |
| Set<GroupInputEdge> groupInputEdges = Sets.newHashSet(); |
| |
| private DAGAccessControls dagAccessControls; |
| Map<String, LocalResource> commonTaskLocalFiles = Maps.newHashMap(); |
| String dagInfo; |
| CallerContext callerContext; |
| private Map<String,String> dagConf = new HashMap<String, String>(); |
| private VertexExecutionContext defaultExecutionContext; |
| |
| private DAG(String name) { |
| this.name = name; |
| } |
| |
| /** |
| * Create a DAG with the specified name. |
| * @param name the name of the DAG |
| * @return this {@link DAG} |
| */ |
| public static DAG create(String name) { |
| return new DAG(name); |
| } |
| |
| /** |
| * Set the files etc that must be provided to the tasks of this DAG |
| * @param localFiles |
| * files that must be available locally for each task. These files |
| * may be regular files, archives etc. as specified by the value |
| * elements of the map. |
| * @return {@link DAG} |
| */ |
| public synchronized DAG addTaskLocalFiles(Map<String, LocalResource> localFiles) { |
| Preconditions.checkNotNull(localFiles); |
| TezCommonUtils.addAdditionalLocalResources(localFiles, commonTaskLocalFiles, "DAG " + getName()); |
| return this; |
| } |
| |
| public synchronized DAG addVertex(Vertex vertex) { |
| if (vertices.containsKey(vertex.getName())) { |
| throw new IllegalStateException( |
| "Vertex " + vertex.getName() + " already defined!"); |
| } |
| vertices.put(vertex.getName(), vertex); |
| return this; |
| } |
| |
| public synchronized Vertex getVertex(String vertexName) { |
| return vertices.get(vertexName); |
| } |
| |
| /** |
| * One of the methods that can be used to provide information about required |
| * Credentials when running on a secure cluster. A combination of this and |
| * addURIsForCredentials should be used to specify information about all |
| * credentials required by a DAG. AM specific credentials are not used when |
| * executing a DAG. |
| * |
| * Set credentials which will be required to run this dag. This method can be |
| * used if the client has already obtained some or all of the required |
| * credentials. |
| * |
| * @param credentials Credentials for the DAG |
| * @return {@link DAG} |
| */ |
| public synchronized DAG setCredentials(Credentials credentials) { |
| this.credentials = credentials; |
| return this; |
| } |
| |
| /** |
| * Set description info for this DAG that can be used for visualization purposes. |
| * @param dagInfo JSON blob as a serialized string. |
| * Recognized keys by the UI are: |
| * "context" - The application context in which this DAG is being used. |
| * For example, this could be set to "Hive" or "Pig" if |
| * this is being run as part of a Hive or Pig script. |
| * "description" - General description on what this DAG is going to do. |
| * In the case of Hive, this could be the SQL query text. |
| * @return {@link DAG} |
| */ |
| @Deprecated |
| public synchronized DAG setDAGInfo(String dagInfo) { |
| Preconditions.checkNotNull(dagInfo); |
| this.dagInfo = dagInfo; |
| return this; |
| } |
| |
| |
| /** |
| * Set the Context in which Tez is being called. |
| * @param callerContext Caller Context |
| * @return {@link DAG} |
| */ |
| public synchronized DAG setCallerContext(CallerContext callerContext) { |
| Preconditions.checkNotNull(callerContext); |
| this.callerContext = callerContext; |
| return this; |
| } |
| |
| /** |
| * Create a group of vertices that share a common output. This can be used to implement |
| * unions efficiently. |
| * @param name Name of the group. |
| * @param members {@link Vertex} members of the group |
| * @return {@link DAG} |
| */ |
| public synchronized VertexGroup createVertexGroup(String name, Vertex... members) { |
| // vertex group name should be unique. |
| VertexGroup uv = new VertexGroup(name, members); |
| if (!vertexGroups.add(uv)) { |
| throw new IllegalStateException( |
| "VertexGroup " + name + " already defined!"); |
| } |
| |
| return uv; |
| } |
| |
| @Private |
| public synchronized Credentials getCredentials() { |
| return this.credentials; |
| } |
| |
| |
| /** |
| * Set Access controls for the DAG. Which user/groups can view the DAG progess/history and |
| * who can modify the DAG i.e. kill the DAG. |
| * The owner of the Tez Session and the user submitting the DAG are super-users and have access |
| * to all operations on the DAG. |
| * @param accessControls Access Controls |
| * @return {@link DAG} |
| */ |
| public synchronized DAG setAccessControls(DAGAccessControls accessControls) { |
| this.dagAccessControls = accessControls; |
| return this; |
| } |
| |
| @Private |
| public synchronized DAGAccessControls getDagAccessControls() { |
| return dagAccessControls; |
| } |
| |
| /** |
| * One of the methods that can be used to provide information about required |
| * Credentials when running on a secure cluster. A combination of this and |
| * setCredentials should be used to specify information about all credentials |
| * required by a DAG. AM specific credentials are not used when executing a |
| * DAG. |
| * |
| * This method can be used to specify a list of URIs for which Credentials |
| * need to be obtained so that the job can run. An incremental list of URIs |
| * can be provided by making multiple calls to the method. |
| * |
| * Currently, @{link credentials} can only be fetched for HDFS and other |
| * {@link org.apache.hadoop.fs.FileSystem} implementations that support |
| * credentials. |
| * |
| * @param uris |
| * a list of {@link URI}s |
| * @return {@link DAG} |
| */ |
| public synchronized DAG addURIsForCredentials(Collection<URI> uris) { |
| Preconditions.checkNotNull(uris, "URIs cannot be null"); |
| urisForCredentials.addAll(uris); |
| return this; |
| } |
| |
| /** |
| * |
| * @return an unmodifiable list representing the URIs for which credentials |
| * are required. |
| */ |
| @Private |
| public synchronized Collection<URI> getURIsForCredentials() { |
| return Collections.unmodifiableCollection(urisForCredentials); |
| } |
| |
| @Private |
| public synchronized Set<Vertex> getVertices() { |
| return Collections.unmodifiableSet(this.vertices.values()); |
| } |
| |
| /** |
| * Add an {@link Edge} connecting vertices in the DAG |
| * |
| * All edges within a DAG must be either named (created via |
| * {@link org.apache.tez.dag.api.Edge#create(Vertex, Vertex, EdgeProperty, String)}) or unnamed |
| * (created via {@link org.apache.tez.dag.api.Edge#create(Vertex, Vertex, EdgeProperty)}). |
| * If edges are named, all inbound edges to a vertex should have unique names. Likewise for outbound edges. |
| * A vertex can have an inbound edge that uses the same name as that used by an outbound edge. |
| * |
| * @param edge The edge to be added |
| * @return {@link DAG} |
| */ |
| public synchronized DAG addEdge(Edge edge) { |
| // Sanity checks |
| if (!vertices.containsValue(edge.getInputVertex())) { |
| throw new IllegalArgumentException( |
| "Input vertex " + edge.getInputVertex() + " doesn't exist!"); |
| } |
| if (!vertices.containsValue(edge.getOutputVertex())) { |
| throw new IllegalArgumentException( |
| "Output vertex " + edge.getOutputVertex() + " doesn't exist!"); |
| } |
| if (edges.contains(edge)) { |
| throw new IllegalArgumentException( |
| "Edge " + edge + " already defined!"); |
| } |
| |
| // inform the vertices |
| edge.getInputVertex().addOutputVertex(edge.getOutputVertex(), edge); |
| edge.getOutputVertex().addInputVertex(edge.getInputVertex(), edge); |
| |
| edges.add(edge); |
| return this; |
| } |
| |
| /** |
| * Add a {@link GroupInputEdge} to the DAG. |
| * @param edge {@link GroupInputEdge} |
| * @return {@link DAG} |
| */ |
| public synchronized DAG addEdge(GroupInputEdge edge) { |
| // Sanity checks |
| if (!vertexGroups.contains(edge.getInputVertexGroup())) { |
| throw new IllegalArgumentException( |
| "Input vertex " + edge.getInputVertexGroup() + " doesn't exist!"); |
| } |
| if (!vertices.containsValue(edge.getOutputVertex())) { |
| throw new IllegalArgumentException( |
| "Output vertex " + edge.getOutputVertex() + " doesn't exist!"); |
| } |
| if (groupInputEdges.contains(edge)) { |
| throw new IllegalArgumentException( |
| "GroupInputEdge " + edge + " already defined!"); |
| } |
| |
| VertexGroup av = edge.getInputVertexGroup(); |
| av.addOutputVertex(edge.getOutputVertex(), edge); |
| groupInputEdges.add(edge); |
| |
| // add new edge between members of VertexGroup and destVertex of the GroupInputEdge |
| List<Edge> newEdges = Lists.newLinkedList(); |
| Vertex dstVertex = edge.getOutputVertex(); |
| VertexGroup uv = edge.getInputVertexGroup(); |
| for (Vertex member : uv.getMembers()) { |
| newEdges.add(Edge.create(member, dstVertex, edge.getEdgeProperty())); |
| } |
| dstVertex.addGroupInput(uv.getGroupName(), uv.getGroupInfo()); |
| |
| for (Edge e : newEdges) { |
| addEdge(e); |
| } |
| |
| return this; |
| } |
| |
| /** |
| * Get the DAG name |
| * @return DAG name |
| */ |
| public String getName() { |
| return this.name; |
| } |
| |
| /** |
| * This is currently used to setup additional configuration parameters which will be available |
| * in the DAG configuration used in the AppMaster. This API would be used for properties which |
| * are used by the Tez framework while executing the DAG. As an example, the number of attempts |
| * for a task.</p> |
| * |
| * A DAG inherits it's base properties from the ApplicationMaster within which it's running. This |
| * method allows for these properties to be overridden. |
| * |
| * Currently, properties which are used by the task runtime, such as the task to AM |
| * heartbeat interval, cannot be changed using this method. </p> |
| * |
| * Note: This API does not add any configuration to runtime components such as InputInitializers, |
| * OutputCommitters, Inputs and Outputs. |
| * |
| * @param property the property name |
| * @param value the value for the property |
| * @return the current DAG being constructed |
| */ |
| @InterfaceStability.Unstable |
| public DAG setConf(String property, String value) { |
| TezConfiguration.validateProperty(property, Scope.DAG); |
| dagConf.put(property, value); |
| return this; |
| } |
| |
| /** |
| * Set history log level for this DAG. This config overrides the default or one set at the session |
| * level. |
| * |
| * @param historyLogLevel The ATS history log level for this DAG. |
| * |
| * @return this DAG |
| */ |
| public DAG setHistoryLogLevel(HistoryLogLevel historyLogLevel) { |
| return this.setConf(TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL, historyLogLevel.name()); |
| } |
| |
| /** |
| * Sets the default execution context for the DAG. This can be overridden at a per Vertex level. |
| * See {@link org.apache.tez.dag.api.Vertex#setExecutionContext(VertexExecutionContext)} |
| * |
| * @param vertexExecutionContext the default execution context for the DAG |
| * |
| * @return this DAG |
| */ |
| @Public |
| @InterfaceStability.Unstable |
| public synchronized DAG setExecutionContext(VertexExecutionContext vertexExecutionContext) { |
| this.defaultExecutionContext = vertexExecutionContext; |
| return this; |
| } |
| |
| @Private |
| VertexExecutionContext getDefaultExecutionContext() { |
| return this.defaultExecutionContext; |
| } |
| |
| @Private |
| @VisibleForTesting |
| public Map<String,String> getDagConf() { |
| return dagConf; |
| } |
| |
| @Private |
| public Map<String, LocalResource> getTaskLocalFiles() { |
| return commonTaskLocalFiles; |
| } |
| |
| @Private |
| @VisibleForTesting |
| void checkAndInferOneToOneParallelism() { |
| // infer all 1-1 via dependencies |
| // collect all 1-1 edges where the source parallelism is set |
| Set<Vertex> newKnownTasksVertices = Sets.newHashSet(); |
| for (Vertex vertex : vertices.values()) { |
| if (vertex.getParallelism() > -1) { |
| newKnownTasksVertices.add(vertex); |
| } |
| } |
| |
| // walk through all known source 1-1 edges and infer parallelism |
| // add newly inferred vertices for consideration as known sources |
| // the outer loop will run for every new level of inferring the parallelism |
| // however, the entire logic will process each vertex only once |
| while(!newKnownTasksVertices.isEmpty()) { |
| Set<Vertex> knownTasksVertices = Sets.newHashSet(newKnownTasksVertices); |
| newKnownTasksVertices.clear(); |
| for (Vertex v : knownTasksVertices) { |
| for (Edge e : v.getOutputEdges()) { |
| if (e.getEdgeProperty().getDataMovementType() == DataMovementType.ONE_TO_ONE) { |
| Vertex outVertex = e.getOutputVertex(); |
| if (outVertex.getParallelism() == -1) { |
| LOG.info("Inferring parallelism for vertex: " |
| + outVertex.getName() + " to be " + v.getParallelism() |
| + " from 1-1 connection with vertex " + v.getName()); |
| outVertex.setParallelism(v.getParallelism()); |
| newKnownTasksVertices.add(outVertex); |
| } |
| } |
| } |
| } |
| } |
| |
| // check for inconsistency and errors |
| for (Edge e : edges) { |
| Vertex inputVertex = e.getInputVertex(); |
| Vertex outputVertex = e.getOutputVertex(); |
| |
| if (e.getEdgeProperty().getDataMovementType() == DataMovementType.ONE_TO_ONE) { |
| if (inputVertex.getParallelism() != outputVertex.getParallelism()) { |
| // both should be equal or equal to -1. |
| if (outputVertex.getParallelism() != -1) { |
| throw new TezUncheckedException( |
| "1-1 Edge. Destination vertex parallelism must match source vertex. " |
| + "Vertex: " + inputVertex.getName() + " does not match vertex: " |
| + outputVertex.getName()); |
| } |
| } |
| } |
| } |
| |
| // check the vertices with -1 parallelism, currently only 3 cases are allowed to has -1 parallelism. |
| // It is OK not using topological order to check vertices here. |
| // 1. has input initializers |
| // 2. 1-1 uninited sources |
| // 3. has custom vertex manager |
| for (Vertex vertex : vertices.values()) { |
| if (vertex.getParallelism() == -1) { |
| boolean hasInputInitializer = false; |
| if (vertex.getDataSources() != null && !vertex.getDataSources().isEmpty()) { |
| for (DataSourceDescriptor ds : vertex.getDataSources()) { |
| if (ds.getInputInitializerDescriptor() != null) { |
| hasInputInitializer = true; |
| break; |
| } |
| } |
| } |
| if (hasInputInitializer) { |
| continue; |
| } else { |
| // Account for the case where the vertex has a data source with a determined number of |
| // shards e.g. splits calculated on the client and not in the AM |
| // In this case, vertex parallelism is setup later using the data source's numShards |
| // and as a result, an initializer is not needed. |
| if (vertex.getDataSources() != null |
| && vertex.getDataSources().size() == 1 |
| && vertex.getDataSources().get(0).getNumberOfShards() > -1) { |
| continue; |
| } |
| } |
| |
| boolean has1to1UninitedSources = false; |
| if (vertex.getInputVertices()!= null && !vertex.getInputVertices().isEmpty()) { |
| for (Vertex srcVertex : vertex.getInputVertices()) { |
| if (srcVertex.getParallelism() == -1) { |
| has1to1UninitedSources = true; |
| break; |
| } |
| } |
| } |
| if (has1to1UninitedSources) { |
| continue; |
| } |
| |
| if (vertex.getVertexManagerPlugin() != null) { |
| continue; |
| } |
| throw new IllegalStateException(vertex.getName() + |
| " has -1 tasks but does not have input initializers, " + |
| "1-1 uninited sources or custom vertex manager to set it at runtime"); |
| } |
| } |
| } |
| |
| // AnnotatedVertex is used by verify() |
| private static class AnnotatedVertex { |
| Vertex v; |
| |
| int index; //for Tarjan's algorithm |
| int lowlink; //for Tarjan's algorithm |
| boolean onstack; //for Tarjan's algorithm |
| |
| |
| private AnnotatedVertex(Vertex v) { |
| this.v = v; |
| index = -1; |
| lowlink = -1; |
| } |
| } |
| |
| // verify() |
| // |
| // Default rules |
| // Illegal: |
| // - duplicate vertex id |
| // - cycles |
| // |
| // Ok: |
| // - orphaned vertex. Occurs in map-only |
| // - islands. Occurs if job has unrelated workflows. |
| // |
| // Not yet categorized: |
| // - orphaned vertex in DAG of >1 vertex. Could be unrelated map-only job. |
| // - v1->v2 via two edges. perhaps some self-join job would use this? |
| // |
| // "restricted" mode: |
| // In short term, the supported DAGs are limited. Call with restricted=true for these verifications. |
| // Illegal: |
| // - any vertex with more than one input or output edge. (n-ary input, n-ary merge) |
| @VisibleForTesting |
| void verify() throws IllegalStateException { |
| verify(true); |
| } |
| |
| @VisibleForTesting |
| Deque<String> verify(boolean restricted) throws IllegalStateException { |
| if (vertices.isEmpty()) { |
| throw new IllegalStateException("Invalid dag containing 0 vertices"); |
| } |
| |
| // check for valid vertices, duplicate vertex names, |
| // and prepare for cycle detection |
| Map<String, AnnotatedVertex> vertexMap = new HashMap<String, AnnotatedVertex>(); |
| for (Vertex v : vertices.values()) { |
| if (vertexMap.containsKey(v.getName())) { |
| throw new IllegalStateException("DAG contains multiple vertices" |
| + " with name: " + v.getName()); |
| } |
| vertexMap.put(v.getName(), new AnnotatedVertex(v)); |
| } |
| |
| // named edge cannot be mixed with unnamed edge or group edge |
| Edge namedEdge = null, unnamedEdge = null; |
| for (Edge e : edges) { |
| if (e.getName() == null) { |
| unnamedEdge = e; |
| } else { |
| namedEdge = e; |
| } |
| |
| if (namedEdge != null && !groupInputEdges.isEmpty()) { |
| throw new IllegalStateException("DAG shouldn't contains both named edge " + namedEdge |
| + " and group edge " + groupInputEdges.iterator().next()); |
| } |
| if (namedEdge != null && unnamedEdge != null) { |
| throw new IllegalStateException("DAG shouldn't contains both named edge " + namedEdge |
| + " and unnamed edge " + unnamedEdge); |
| } |
| } |
| |
| Map<Vertex, List<Edge>> inEdgeMap = new HashMap<>(); |
| Map<Vertex, List<Edge>> outEdgeMap = new HashMap<>(); |
| for (Edge e : edges) { |
| // Construct structure for cycle detection |
| Vertex inputVertex = e.getInputVertex(); |
| Vertex outputVertex = e.getOutputVertex(); |
| |
| List<Edge> outEdgeList = outEdgeMap.get(inputVertex); |
| if (outEdgeList == null) { |
| outEdgeList = new ArrayList<Edge>(); |
| outEdgeMap.put(inputVertex, outEdgeList); |
| } |
| outEdgeList.add(e); |
| |
| List<Edge> inEdgeList = inEdgeMap.get(outputVertex); |
| if (inEdgeList == null) { |
| inEdgeList = new ArrayList<Edge>(); |
| inEdgeMap.put(outputVertex, inEdgeList); |
| } |
| inEdgeList.add(e); |
| } |
| |
| // check input and output names don't collide with vertex names |
| for (Vertex vertex : vertices.values()) { |
| for (RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor> |
| input : vertex.getInputs()) { |
| if (vertexMap.containsKey(input.getName())) { |
| throw new IllegalStateException("Vertex: " |
| + vertex.getName() |
| + " contains an Input with the same name as vertex: " |
| + input.getName()); |
| } |
| } |
| for (RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> |
| output : vertex.getOutputs()) { |
| if (vertexMap.containsKey(output.getName())) { |
| throw new IllegalStateException("Vertex: " |
| + vertex.getName() |
| + " contains an Output with the same name as vertex: " |
| + output.getName()); |
| } |
| } |
| } |
| |
| // Check for valid InputNames |
| for (Entry<Vertex, List<Edge>> entry : inEdgeMap.entrySet()) { |
| Vertex vertex = entry.getKey(); |
| Set<String> inputs = new HashSet<>(); |
| |
| for (Edge edge : entry.getValue()) { |
| String name = edge.getName(); |
| if (name == null) { |
| name = edge.getInputVertex().getName(); |
| } |
| if (inputs.contains(name)) { |
| throw new IllegalStateException("Vertex: " + vertex.getName() + " contains multiple " + |
| "incoming edges with name " + name); |
| } |
| inputs.add(name); |
| } |
| for (RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor> |
| input : vertex.getInputs()) { |
| if (inputs.contains(input.getName())) { |
| throw new IllegalStateException("Vertex: " |
| + vertex.getName() |
| + " contains an incoming " + (namedEdge != null ? "edge" : "vertex") |
| + " and Input with the same name: " + input.getName()); |
| } |
| } |
| } |
| |
| for (Entry<Vertex, List<Edge>> entry : outEdgeMap.entrySet()) { |
| Vertex vertex = entry.getKey(); |
| Set<String> outputs = new HashSet<>(); |
| |
| for (Edge edge : entry.getValue()) { |
| String name = edge.getName(); |
| if (name == null) { |
| name = edge.getOutputVertex().getName(); |
| } |
| if (outputs.contains(name)) { |
| throw new IllegalStateException("Vertex: " + vertex.getName() + " contains multiple " + |
| "outgoing edges with name " + name); |
| } |
| outputs.add(name); |
| } |
| for (RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> |
| output : vertex.getOutputs()) { |
| if (outputs.contains(output.getName())) { |
| throw new IllegalStateException("Vertex: " |
| + vertex.getName() |
| + " contains an outgoing " + (namedEdge != null ? "edge" : "vertex") |
| + " and Output with the same name: " + output.getName()); |
| } |
| } |
| } |
| |
| |
| // Not checking for repeated input names / output names vertex names on the same vertex, |
| // since we only allow 1 at the moment. |
| // When additional inputs are supported, this can be chceked easily (and early) |
| // within the addInput / addOutput call itself. |
| |
| Deque<String> topologicalVertexStack = detectCycles(outEdgeMap, vertexMap); |
| |
| checkAndInferOneToOneParallelism(); |
| |
| if (restricted) { |
| for (Edge e : edges) { |
| if (e.getEdgeProperty().getDataSourceType() != |
| DataSourceType.PERSISTED) { |
| throw new IllegalStateException( |
| "Unsupported source type on edge. " + e); |
| } |
| if (e.getEdgeProperty().getSchedulingType() != |
| SchedulingType.SEQUENTIAL) { |
| throw new IllegalStateException( |
| "Unsupported scheduling type on edge. " + e); |
| } |
| } |
| } |
| |
| // check for conflicts between dag level local resource and vertex level local resource |
| for (Vertex v : vertices.values()) { |
| for (Map.Entry<String, LocalResource> localResource : v.getTaskLocalFiles().entrySet()) { |
| String resourceName = localResource.getKey(); |
| LocalResource resource = localResource.getValue(); |
| if (commonTaskLocalFiles.containsKey(resourceName) |
| && !commonTaskLocalFiles.get(resourceName).equals(resource)) { |
| throw new IllegalStateException("There is conflicting local resource (" + resourceName |
| + ") between dag local resource and vertex " + v.getName() + " local resource. " |
| + "\nResource of dag : " + commonTaskLocalFiles.get(resourceName) |
| + "\nResource of vertex: " + resource); |
| } |
| } |
| } |
| |
| return topologicalVertexStack; |
| } |
| |
| // Adaptation of Tarjan's algorithm for connected components. |
| // http://en.wikipedia.org/wiki/Tarjan%27s_strongly_connected_components_algorithm |
| private Deque<String> detectCycles(Map<Vertex, List<Edge>> edgeMap, |
| Map<String, AnnotatedVertex> vertexMap) |
| throws IllegalStateException { |
| Deque<String> topologicalVertexStack = new LinkedList<String>(); |
| Integer nextIndex = 0; // boxed integer so it is passed by reference. |
| Stack<AnnotatedVertex> stack = new Stack<DAG.AnnotatedVertex>(); |
| for (AnnotatedVertex av : vertexMap.values()) { |
| if (av.index == -1) { |
| assert stack.empty(); |
| strongConnect(av, vertexMap, edgeMap, stack, nextIndex, topologicalVertexStack); |
| } |
| } |
| return topologicalVertexStack; |
| } |
| |
| // part of Tarjan's algorithm for connected components. |
| // http://en.wikipedia.org/wiki/Tarjan%27s_strongly_connected_components_algorithm |
| private void strongConnect( |
| AnnotatedVertex av, |
| Map<String, AnnotatedVertex> vertexMap, |
| Map<Vertex, List<Edge>> edgeMap, |
| Stack<AnnotatedVertex> stack, |
| Integer nextIndex, |
| Deque<String> topologicalVertexStack) throws IllegalStateException { |
| av.index = nextIndex; |
| av.lowlink = nextIndex; |
| nextIndex++; |
| stack.push(av); |
| av.onstack = true; |
| |
| List<Edge> edges = edgeMap.get(av.v); |
| if (edges != null) { |
| for (Edge e : edgeMap.get(av.v)) { |
| AnnotatedVertex outVertex = vertexMap.get(e.getOutputVertex().getName()); |
| if (outVertex.index == -1) { |
| strongConnect(outVertex, vertexMap, edgeMap, stack, nextIndex, topologicalVertexStack); |
| av.lowlink = Math.min(av.lowlink, outVertex.lowlink); |
| } else if (outVertex.onstack) { |
| // strongly connected component detected, but we will wait till later so that the full cycle can be displayed. |
| // update lowlink in case outputVertex should be considered the root of this component. |
| av.lowlink = Math.min(av.lowlink, outVertex.index); |
| } |
| } |
| } |
| |
| if (av.lowlink == av.index) { |
| AnnotatedVertex pop = stack.pop(); |
| pop.onstack = false; |
| if (pop != av) { |
| // there was something on the stack other than this "av". |
| // this indicates there is a scc/cycle. It comprises all nodes from top of stack to "av" |
| StringBuilder message = new StringBuilder(); |
| message.append(av.v.getName()).append(" <- "); |
| for (; pop != av; pop = stack.pop()) { |
| message.append(pop.v.getName()).append(" <- "); |
| pop.onstack = false; |
| } |
| message.append(av.v.getName()); |
| throw new IllegalStateException("DAG contains a cycle: " + message); |
| } else { |
| // detect self-cycle |
| if (edgeMap.containsKey(pop.v)) { |
| for (Edge edge : edgeMap.get(pop.v)) { |
| if (edge.getOutputVertex().equals(pop.v)) { |
| throw new IllegalStateException("DAG contains a self-cycle on vertex:" + pop.v.getName()); |
| } |
| } |
| } |
| } |
| topologicalVertexStack.push(av.v.getName()); |
| } |
| } |
| |
| // create protobuf message describing DAG |
| @Private |
| public DAGPlan createDag(Configuration tezConf, Credentials extraCredentials, |
| Map<String, LocalResource> tezJarResources, LocalResource binaryConfig, |
| boolean tezLrsAsArchive) { |
| return createDag(tezConf, extraCredentials, tezJarResources, binaryConfig, tezLrsAsArchive, |
| null, null); |
| } |
| |
| // create protobuf message describing DAG |
| @Private |
| public synchronized DAGPlan createDag(Configuration tezConf, Credentials extraCredentials, |
| Map<String, LocalResource> tezJarResources, LocalResource binaryConfig, |
| boolean tezLrsAsArchive, ServicePluginsDescriptor servicePluginsDescriptor, |
| JavaOptsChecker javaOptsChecker) { |
| Deque<String> topologicalVertexStack = verify(true); |
| |
| DAGPlan.Builder dagBuilder = DAGPlan.newBuilder(); |
| dagBuilder.setName(this.name); |
| |
| if (this.callerContext != null) { |
| dagBuilder.setCallerContext(DagTypeConverters.convertCallerContextToProto(callerContext)); |
| } |
| if (this.dagInfo != null && !this.dagInfo.isEmpty()) { |
| dagBuilder.setDagInfo(this.dagInfo); |
| } |
| |
| // Setup default execution context. |
| VertexExecutionContext defaultContext = getDefaultExecutionContext(); |
| verifyExecutionContext(defaultContext, servicePluginsDescriptor, "DAGDefault"); |
| if (defaultContext != null) { |
| DAGProtos.VertexExecutionContextProto contextProto = DagTypeConverters.convertToProto( |
| defaultContext); |
| dagBuilder.setDefaultExecutionContext(contextProto); |
| } |
| |
| if (!vertexGroups.isEmpty()) { |
| for (VertexGroup av : vertexGroups) { |
| GroupInfo groupInfo = av.getGroupInfo(); |
| PlanVertexGroupInfo.Builder groupBuilder = PlanVertexGroupInfo.newBuilder(); |
| groupBuilder.setGroupName(groupInfo.getGroupName()); |
| for (Vertex v : groupInfo.getMembers()) { |
| groupBuilder.addGroupMembers(v.getName()); |
| } |
| groupBuilder.addAllOutputs(groupInfo.outputs); |
| for (Map.Entry<String, InputDescriptor> entry : |
| groupInfo.edgeMergedInputs.entrySet()) { |
| groupBuilder.addEdgeMergedInputs( |
| PlanGroupInputEdgeInfo.newBuilder().setDestVertexName(entry.getKey()). |
| setMergedInput(DagTypeConverters.convertToDAGPlan(entry.getValue()))); |
| } |
| dagBuilder.addVertexGroups(groupBuilder); |
| } |
| } |
| |
| Credentials dagCredentials = new Credentials(); |
| if (extraCredentials != null) { |
| dagCredentials.mergeAll(extraCredentials); |
| } |
| dagCredentials.mergeAll(credentials); |
| if (!commonTaskLocalFiles.isEmpty()) { |
| dagBuilder.addAllLocalResource(DagTypeConverters.convertToDAGPlan(commonTaskLocalFiles)); |
| } |
| |
| Preconditions.checkArgument(topologicalVertexStack.size() == vertices.size(), |
| "size of topologicalVertexStack is:" + topologicalVertexStack.size() + |
| " while size of vertices is:" + vertices.size() + |
| ", make sure they are the same in order to sort the vertices"); |
| while(!topologicalVertexStack.isEmpty()) { |
| Vertex vertex = vertices.get(topologicalVertexStack.pop()); |
| // infer credentials, resources and parallelism from data source |
| Resource vertexTaskResource = vertex.getTaskResource(); |
| if (vertexTaskResource == null) { |
| vertexTaskResource = Resource.newInstance(tezConf.getInt( |
| TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, |
| TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB_DEFAULT), tezConf.getInt( |
| TezConfiguration.TEZ_TASK_RESOURCE_CPU_VCORES, |
| TezConfiguration.TEZ_TASK_RESOURCE_CPU_VCORES_DEFAULT)); |
| } |
| Map<String, LocalResource> vertexLRs = Maps.newHashMap(); |
| vertexLRs.putAll(vertex.getTaskLocalFiles()); |
| List<DataSourceDescriptor> dataSources = vertex.getDataSources(); |
| for (DataSourceDescriptor dataSource : dataSources) { |
| if (dataSource.getCredentials() != null) { |
| dagCredentials.addAll(dataSource.getCredentials()); |
| } |
| if (dataSource.getAdditionalLocalFiles() != null) { |
| TezCommonUtils |
| .addAdditionalLocalResources(dataSource.getAdditionalLocalFiles(), vertexLRs, |
| "Vertex " + vertex.getName()); |
| } |
| } |
| if (tezJarResources != null) { |
| TezCommonUtils |
| .addAdditionalLocalResources(tezJarResources, vertexLRs, "Vertex " + vertex.getName()); |
| } |
| if (binaryConfig != null) { |
| vertexLRs.put(TezConstants.TEZ_PB_BINARY_CONF_NAME, binaryConfig); |
| } |
| int vertexParallelism = vertex.getParallelism(); |
| VertexLocationHint vertexLocationHint = vertex.getLocationHint(); |
| if (dataSources.size() == 1) { |
| DataSourceDescriptor dataSource = dataSources.get(0); |
| if (vertexParallelism == -1 && dataSource.getNumberOfShards() > -1) { |
| vertexParallelism = dataSource.getNumberOfShards(); |
| } |
| if (vertexLocationHint == null && dataSource.getLocationHint() != null) { |
| vertexLocationHint = dataSource.getLocationHint(); |
| } |
| } |
| if (vertexParallelism == -1) { |
| Preconditions.checkState(vertexLocationHint == null, |
| "Cannot specify vertex location hint without specifying vertex parallelism. Vertex: " |
| + vertex.getName()); |
| } else if (vertexLocationHint != null) { |
| Preconditions.checkState(vertexParallelism == vertexLocationHint.getTaskLocationHints().size(), |
| "vertex task location hint must equal vertex parallelism. Vertex: " + vertex.getName()); |
| } |
| for (DataSinkDescriptor dataSink : vertex.getDataSinks()) { |
| if (dataSink.getCredentials() != null) { |
| dagCredentials.addAll(dataSink.getCredentials()); |
| } |
| } |
| |
| VertexPlan.Builder vertexBuilder = VertexPlan.newBuilder(); |
| vertexBuilder.setName(vertex.getName()); |
| vertexBuilder.setType(PlanVertexType.NORMAL); // vertex type is implicitly NORMAL until TEZ-46. |
| vertexBuilder.setProcessorDescriptor(DagTypeConverters |
| .convertToDAGPlan(vertex.getProcessorDescriptor())); |
| |
| // Vertex ExecutionContext setup |
| VertexExecutionContext execContext = vertex.getVertexExecutionContext(); |
| verifyExecutionContext(execContext, servicePluginsDescriptor, vertex.getName()); |
| if (execContext != null) { |
| DAGProtos.VertexExecutionContextProto contextProto = |
| DagTypeConverters.convertToProto(execContext); |
| vertexBuilder.setExecutionContext(contextProto); |
| } |
| // End of VertexExecutionContext setup. |
| |
| if (vertex.getInputs().size() > 0) { |
| for (RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor> input : vertex.getInputs()) { |
| vertexBuilder.addInputs(DagTypeConverters.convertToDAGPlan(input)); |
| } |
| } |
| if (vertex.getOutputs().size() > 0) { |
| for (RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> output : vertex.getOutputs()) { |
| vertexBuilder.addOutputs(DagTypeConverters.convertToDAGPlan(output)); |
| } |
| } |
| |
| if (vertex.getConf()!= null && vertex.getConf().size() > 0) { |
| ConfigurationProto.Builder confBuilder = ConfigurationProto.newBuilder(); |
| for (Map.Entry<String, String> entry : vertex.getConf().entrySet()) { |
| PlanKeyValuePair.Builder keyValueBuilder = PlanKeyValuePair.newBuilder(); |
| keyValueBuilder.setKey(entry.getKey()); |
| keyValueBuilder.setValue(entry.getValue()); |
| confBuilder.addConfKeyValues(keyValueBuilder); |
| } |
| vertexBuilder.setVertexConf(confBuilder); |
| } |
| |
| //task config |
| PlanTaskConfiguration.Builder taskConfigBuilder = PlanTaskConfiguration.newBuilder(); |
| taskConfigBuilder.setNumTasks(vertexParallelism); |
| taskConfigBuilder.setMemoryMb(vertexTaskResource.getMemory()); |
| taskConfigBuilder.setVirtualCores(vertexTaskResource.getVirtualCores()); |
| |
| try { |
| taskConfigBuilder.setJavaOpts( |
| TezClientUtils.addDefaultsToTaskLaunchCmdOpts(vertex.getTaskLaunchCmdOpts(), tezConf, |
| javaOptsChecker)); |
| } catch (TezException e) { |
| throw new TezUncheckedException("Invalid TaskLaunchCmdOpts defined for Vertex " |
| + vertex.getName() + " : " + e.getMessage(), e); |
| } |
| |
| taskConfigBuilder.setTaskModule(vertex.getName()); |
| if (!vertexLRs.isEmpty()) { |
| taskConfigBuilder.addAllLocalResource(DagTypeConverters.convertToDAGPlan(vertexLRs)); |
| } |
| |
| Map<String, String> taskEnv = Maps.newHashMap(vertex.getTaskEnvironment()); |
| TezYARNUtils.setupDefaultEnv(taskEnv, tezConf, |
| TezConfiguration.TEZ_TASK_LAUNCH_ENV, |
| TezConfiguration.TEZ_TASK_LAUNCH_ENV_DEFAULT, |
| TezConfiguration.TEZ_TASK_LAUNCH_CLUSTER_DEFAULT_ENV, |
| TezConfiguration.TEZ_TASK_LAUNCH_CLUSTER_DEFAULT_ENV_DEFAULT, |
| tezLrsAsArchive); |
| for (Map.Entry<String, String> entry : taskEnv.entrySet()) { |
| PlanKeyValuePair.Builder envSettingBuilder = PlanKeyValuePair.newBuilder(); |
| envSettingBuilder.setKey(entry.getKey()); |
| envSettingBuilder.setValue(entry.getValue()); |
| taskConfigBuilder.addEnvironmentSetting(envSettingBuilder); |
| } |
| |
| if (vertexLocationHint != null) { |
| if (vertexLocationHint.getTaskLocationHints() != null) { |
| for (TaskLocationHint hint : vertexLocationHint.getTaskLocationHints()) { |
| PlanTaskLocationHint.Builder taskLocationHintBuilder = PlanTaskLocationHint.newBuilder(); |
| // we can allow this later on if needed |
| if (hint.getAffinitizedTask() != null) { |
| throw new TezUncheckedException( |
| "Task based affinity may not be specified via the DAG API"); |
| } |
| |
| if (hint.getHosts() != null) { |
| taskLocationHintBuilder.addAllHost(hint.getHosts()); |
| } |
| if (hint.getRacks() != null) { |
| taskLocationHintBuilder.addAllRack(hint.getRacks()); |
| } |
| |
| vertexBuilder.addTaskLocationHint(taskLocationHintBuilder); |
| } |
| } |
| } |
| |
| if (vertex.getVertexManagerPlugin() != null) { |
| vertexBuilder.setVertexManagerPlugin(DagTypeConverters |
| .convertToDAGPlan(vertex.getVertexManagerPlugin())); |
| } |
| |
| for (Edge inEdge : vertex.getInputEdges()) { |
| vertexBuilder.addInEdgeId(inEdge.getId()); |
| } |
| |
| for (Edge outEdge : vertex.getOutputEdges()) { |
| vertexBuilder.addOutEdgeId(outEdge.getId()); |
| } |
| |
| vertexBuilder.setTaskConfig(taskConfigBuilder); |
| dagBuilder.addVertex(vertexBuilder); |
| } |
| |
| for (Edge edge : edges) { |
| EdgePlan.Builder edgeBuilder = EdgePlan.newBuilder(); |
| edgeBuilder.setId(edge.getId()); |
| edgeBuilder.setInputVertexName(edge.getInputVertex().getName()); |
| edgeBuilder.setOutputVertexName(edge.getOutputVertex().getName()); |
| edgeBuilder.setDataMovementType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getDataMovementType())); |
| edgeBuilder.setDataSourceType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getDataSourceType())); |
| edgeBuilder.setSchedulingType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getSchedulingType())); |
| edgeBuilder.setEdgeSource(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getEdgeSource())); |
| edgeBuilder.setEdgeDestination(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getEdgeDestination())); |
| if (edge.getEdgeProperty().getDataMovementType() == DataMovementType.CUSTOM) { |
| if (edge.getEdgeProperty().getEdgeManagerDescriptor() != null) { |
| edgeBuilder.setEdgeManager(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getEdgeManagerDescriptor())); |
| } // else the AM will deal with this. |
| } |
| dagBuilder.addEdge(edgeBuilder); |
| } |
| |
| if (dagAccessControls != null) { |
| dagBuilder.setAclInfo(DagTypeConverters.convertDAGAccessControlsToProto(dagAccessControls)); |
| } |
| |
| ConfigurationProto.Builder confProtoBuilder = ConfigurationProto.newBuilder(); |
| if (!this.dagConf.isEmpty()) { |
| for (Entry<String, String> entry : this.dagConf.entrySet()) { |
| PlanKeyValuePair.Builder kvp = PlanKeyValuePair.newBuilder(); |
| kvp.setKey(entry.getKey()); |
| kvp.setValue(entry.getValue()); |
| confProtoBuilder.addConfKeyValues(kvp); |
| } |
| } |
| // Copy historyLogLevel from tezConf into dagConf if its not overridden in dagConf. |
| String logLevel = this.dagConf.get(TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL); |
| if (logLevel != null) { |
| // The config is from dagConf, we have already added it to the proto above, just check if |
| // the value is valid. |
| if (!HistoryLogLevel.validateLogLevel(logLevel)) { |
| throw new IllegalArgumentException( |
| "Config: " + TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL + |
| " is set to invalid value: " + logLevel); |
| } |
| } else { |
| // Validate and set value from tezConf. |
| logLevel = tezConf.get(TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL); |
| if (logLevel != null) { |
| if (!HistoryLogLevel.validateLogLevel(logLevel)) { |
| throw new IllegalArgumentException( |
| "Config: " + TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL + |
| " is set to invalid value: " + logLevel); |
| } |
| PlanKeyValuePair.Builder kvp = PlanKeyValuePair.newBuilder(); |
| kvp.setKey(TezConfiguration.TEZ_HISTORY_LOGGING_LOGLEVEL); |
| kvp.setValue(logLevel); |
| confProtoBuilder.addConfKeyValues(kvp); |
| } |
| } |
| dagBuilder.setDagConf(confProtoBuilder); |
| |
| if (dagCredentials != null) { |
| dagBuilder.setCredentialsBinary(DagTypeConverters.convertCredentialsToProto(dagCredentials)); |
| TezCommonUtils.logCredentials(LOG, dagCredentials, "dag"); |
| } |
| |
| return dagBuilder.build(); |
| } |
| |
| private void verifyExecutionContext(VertexExecutionContext executionContext, |
| ServicePluginsDescriptor servicePluginsDescriptor, |
| String context) { |
| if (executionContext != null) { |
| if (executionContext.shouldExecuteInContainers()) { |
| if (servicePluginsDescriptor == null || !servicePluginsDescriptor.areContainersEnabled()) { |
| throw new IllegalStateException("Invalid configuration. ExecutionContext for " + context + |
| " specifies container execution but this is disabled in the ServicePluginDescriptor"); |
| } |
| } |
| if (executionContext.shouldExecuteInAm()) { |
| if (servicePluginsDescriptor == null || !servicePluginsDescriptor.isUberEnabled()) { |
| throw new IllegalStateException("Invalid configuration. ExecutionContext for " + context + |
| " specifies AM execution but this is disabled in the ServicePluginDescriptor"); |
| } |
| } |
| if (executionContext.getTaskSchedulerName() != null) { |
| boolean found = false; |
| if (servicePluginsDescriptor != null) { |
| found = checkNamedEntityExists(executionContext.getTaskSchedulerName(), |
| servicePluginsDescriptor.getTaskSchedulerDescriptors()); |
| } |
| if (!found) { |
| throw new IllegalStateException("Invalid configuration. ExecutionContext for " + context + |
| " specifies task scheduler as " + executionContext.getTaskSchedulerName() + |
| " which is not part of the ServicePluginDescriptor"); |
| } |
| } |
| if (executionContext.getContainerLauncherName() != null) { |
| boolean found = false; |
| if (servicePluginsDescriptor != null) { |
| found = checkNamedEntityExists(executionContext.getContainerLauncherName(), |
| servicePluginsDescriptor.getContainerLauncherDescriptors()); |
| } |
| if (!found) { |
| throw new IllegalStateException("Invalid configuration. ExecutionContext for " + context + |
| " specifies container launcher as " + executionContext.getContainerLauncherName() + |
| " which is not part of the ServicePluginDescriptor"); |
| } |
| } |
| if (executionContext.getTaskCommName() != null) { |
| boolean found = false; |
| if (servicePluginsDescriptor != null) { |
| found = checkNamedEntityExists(executionContext.getTaskCommName(), |
| servicePluginsDescriptor.getTaskCommunicatorDescriptors()); |
| } |
| if (!found) { |
| throw new IllegalStateException("Invalid configuration. ExecutionContext for " + context + |
| " specifies task communicator as " + executionContext.getTaskCommName() + |
| " which is not part of the ServicePluginDescriptor"); |
| } |
| } |
| } |
| } |
| |
| private boolean checkNamedEntityExists(String expected, NamedEntityDescriptor[] namedEntities) { |
| if (namedEntities == null) { |
| return false; |
| } |
| for (NamedEntityDescriptor named : namedEntities) { |
| if (named.getEntityName().equals(expected)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| public synchronized CallerContext getCallerContext() { |
| return this.callerContext; |
| } |
| |
| } |
