LogicalPlan implements Serializable, DAG { @Override public <T extends Module> T addModule(@Nonnull String name, T module) { checkArgument(!isNullOrEmpty(name), "module name is null or empty"); if (modules.containsKey(name)) { if (modules.get(name).module == module) { return module; } throw new IllegalArgumentException("duplicate module name: " + modules.get(name)); } if (operators.containsKey(name)) { throw new IllegalArgumentException("duplicate module name: " + modules.get(name)); } ModuleMeta meta = new ModuleMeta(name, module); modules.put(name, meta); return module; } LogicalPlan(); @Override Attribute.AttributeMap getAttributes(); @Override T getValue(Attribute<T> key); @Override void setCounters(Object counters); @Override void sendMetrics(Collection<String> metricNames); @Override T addOperator(@Nonnull String name, Class<T> clazz); @Override T addOperator(@Nonnull String name, T operator); @Override T addModule(@Nonnull String name, T module); @Override T addModule(@Nonnull String name, Class<T> clazz); void removeOperator(Operator operator); @Override StreamMeta addStream(@Nonnull String id); @Override @SuppressWarnings("unchecked") StreamMeta addStream(@Nonnull String id, Operator.OutputPort<? extends T> source, Operator.InputPort<? super T>... sinks); void applyStreamLinks(); @Override @SuppressWarnings("unchecked") StreamMeta addStream(String id, Operator.OutputPort<? extends T> source, Operator.InputPort<? super T> sink1); @Override @SuppressWarnings("unchecked") StreamMeta addStream(String id, Operator.OutputPort<? extends T> source, Operator.InputPort<? super T> sink1, Operator.InputPort<? super T> sink2); StreamMeta getStream(String id); Attribute.AttributeMap getContextAttributes(Operator operator); @Override void setAttribute(Attribute<T> key, T value); @Override void setAttribute(Operator operator, Attribute<T> key, T value); @Override void setOperatorAttribute(Operator operator, Attribute<T> key, T value); @Override void setOutputPortAttribute(Operator.OutputPort<?> port, Attribute<T> key, T value); @Override void setUnifierAttribute(Operator.OutputPort<?> port, Attribute<T> key, T value); @Override void setInputPortAttribute(Operator.InputPort<?> port, Attribute<T> key, T value); List<OperatorMeta> getRootOperators(); @Override List<OperatorMeta> getRootOperatorsMeta(); List<OperatorMeta> getLeafOperators(); Collection<OperatorMeta> getAllOperators(); @Override Collection<OperatorMeta> getAllOperatorsMeta(); Collection<ModuleMeta> getAllModules(); Collection<StreamMeta> getAllStreams(); @Override Collection<StreamMeta> getAllStreamsMeta(); @Override OperatorMeta getOperatorMeta(String operatorName); @Override OperatorMeta getMeta(Operator operator); int getMaxContainerCount(); boolean isDebug(); int getMasterMemoryMB(); String getMasterJVMOptions(); String assertAppPath(); Set<String> getClassNames(); void resetNIndex(); void validate(); String getHostLocality(HashMap<String, Set<String>> nodeAffinities, String operator, HashMap<String, String> hostNamesMapping); void combineSets(HashMap<String, Set<String>> containerAffinities, OperatorPair pair); void convertRegexToList(List<String> operatorNames, AffinityRule rule); void findStronglyConnected(OperatorMeta om, ValidationContext ctx); void findInvalidDelays(OperatorMeta om, List<List<String>> invalidDelays, Stack<OperatorMeta> stack); static void write(DAG dag, OutputStream os); static LogicalPlan read(InputStream is); static Type getPortType(Field f); @Override String toString(); @Override boolean equals(Object o); @Override int hashCode(); static final Attribute<Boolean> IS_CONNECTED_TO_DELAY_OPERATOR; static final String SER_FILE_NAME; static final String LAUNCH_CONFIG_FILE_NAME; static final String MODULE_NAMESPACE_SEPARATOR; static String SUBDIR_CHECKPOINTS; static String SUBDIR_STATS; static String SUBDIR_EVENTS; static Attribute<Boolean> FAST_PUBLISHER_SUBSCRIBER; static Attribute<Long> HDFS_TOKEN_RENEWAL_INTERVAL; static Attribute<Long> HDFS_TOKEN_LIFE_TIME; static Attribute<Long> RM_TOKEN_RENEWAL_INTERVAL; static Attribute<Long> RM_TOKEN_LIFE_TIME; static Attribute<String> PRINCIPAL; static Attribute<String> KEY_TAB_FILE; static Attribute<Double> TOKEN_REFRESH_ANTICIPATORY_FACTOR; static Attribute<String> ARCHIVES; static Attribute<String> FILES; static Attribute<Integer> CONTAINERS_MAX_COUNT; static Attribute<Integer> APPLICATION_ATTEMPT_ID; final Map<String, ModuleMeta> modules; }
@Test(expected = IllegalArgumentException.class) public void testEmptyModuleName() { Module testModule = mock(Module.class); dag.addModule("", testModule); }
@Test(expected = IllegalArgumentException.class) public void testNullModuleName() { Module testModule = mock(Module.class); dag.addModule(null, testModule); }