LogicalPlanConfiguration { public void populateDAG(LogicalPlan dag) { Configuration pconf = new Configuration(conf); for (final String propertyName : this.properties.stringPropertyNames()) { String propertyValue = this.properties.getProperty(propertyName); pconf.setIfUnset(propertyName, propertyValue); } AppConf appConf = this.stramConf.getChild(WILDCARD, StramElement.APPLICATION); if (appConf == null) { LOG.warn("Application configuration not found. Probably an empty app."); return; } Map<String, OperatorConf> operators = appConf.getChildren(StramElement.OPERATOR); Map<OperatorConf, GenericOperator> nodeMap = Maps.newHashMapWithExpectedSize(operators.size()); for (Map.Entry<String, OperatorConf> nodeConfEntry : operators.entrySet()) { OperatorConf nodeConf = nodeConfEntry.getValue(); if (!WILDCARD.equals(nodeConf.id)) { Class<? extends GenericOperator> nodeClass = StramUtils.classForName(nodeConf.getClassNameReqd(), GenericOperator.class); String optJson = nodeConf.getProperties().get(nodeClass.getName()); GenericOperator operator = null; try { if (optJson != null) { ObjectMapper mapper = ObjectMapperFactory.getOperatorValueDeserializer(); operator = mapper.readValue("{\"" + nodeClass.getName() + "\":" + optJson + "}", nodeClass); addOperator(dag, nodeConfEntry.getKey(), operator); } else { operator = addOperator(dag, nodeConfEntry.getKey(), nodeClass); } setOperatorProperties(operator, nodeConf.getProperties()); } catch (IOException e) { throw new IllegalArgumentException("Error setting operator properties " + e.getMessage(), e); } nodeMap.put(nodeConf, operator); } } Map<String, StreamConf> streams = appConf.getChildren(StramElement.STREAM); for (Map.Entry<String, StreamConf> streamConfEntry : streams.entrySet()) { StreamConf streamConf = streamConfEntry.getValue(); DAG.StreamMeta sd = dag.addStream(streamConfEntry.getKey()); sd.setLocality(streamConf.getLocality()); String schemaClassName = streamConf.properties.getProperty(STREAM_SCHEMA); Class<?> schemaClass = null; if (schemaClassName != null) { schemaClass = StramUtils.classForName(schemaClassName, Object.class); } if (streamConf.sourceNode != null) { String portName = null; for (Map.Entry<String, StreamConf> e : streamConf.sourceNode.outputs.entrySet()) { if (e.getValue() == streamConf) { portName = e.getKey(); } } GenericOperator sourceDecl = nodeMap.get(streamConf.sourceNode); Operators.PortMappingDescriptor sourcePortMap = new Operators.PortMappingDescriptor(); Operators.describe(sourceDecl, sourcePortMap); sd.setSource(sourcePortMap.outputPorts.get(portName).component); if (schemaClass != null) { dag.setOutputPortAttribute(sourcePortMap.outputPorts.get(portName).component, PortContext.TUPLE_CLASS, schemaClass); } } for (OperatorConf targetNode : streamConf.targetNodes) { String portName = null; for (Map.Entry<String, StreamConf> e : targetNode.inputs.entrySet()) { if (e.getValue() == streamConf) { portName = e.getKey(); } } GenericOperator targetDecl = nodeMap.get(targetNode); Operators.PortMappingDescriptor targetPortMap = new Operators.PortMappingDescriptor(); Operators.describe(targetDecl, targetPortMap); sd.addSink(targetPortMap.inputPorts.get(portName).component); if (schemaClass != null) { dag.setInputPortAttribute(targetPortMap.inputPorts.get(portName).component, PortContext.TUPLE_CLASS, schemaClass); } } } } LogicalPlanConfiguration(Configuration conf); final void addFromConfiguration(Configuration conf); String getAppAlias(String appPath); LogicalPlanConfiguration addFromJson(JSONObject json, Configuration conf); LogicalPlanConfiguration addFromProperties(Properties props, Configuration conf); Properties getProperties(); Map<String, String> getAppAliases(); LogicalPlan createFromProperties(Properties props, String appName); LogicalPlan createFromJson(JSONObject json, String appName); LogicalPlan createEmptyForRecovery(String appName); LogicalPlan createFromStreamingApplication(StreamingApplication app, String appName); void populateDAG(LogicalPlan dag); void prepareDAG(LogicalPlan dag, StreamingApplication app, String name); static Properties readProperties(String filePath); Map<String, String> getProperties(OperatorMeta ow, String appName); static GenericOperator setOperatorProperties(GenericOperator operator, Map<String, String> properties); static StreamingApplication setApplicationProperties(StreamingApplication application, Map<String, String> properties); static BeanMap getObjectProperties(Object obj); void setOperatorProperties(LogicalPlan dag, String applicationName); void setModuleProperties(LogicalPlan dag, String applicationName); void setApplicationConfiguration(final LogicalPlan dag, String appName, StreamingApplication app); static final String GATEWAY_PREFIX; static final String GATEWAY_LISTEN_ADDRESS; static final String STREAM_PREFIX; static final String STREAM_SOURCE; static final String STREAM_SINKS; static final String STREAM_TEMPLATE; static final String STREAM_LOCALITY; static final String STREAM_SCHEMA; static final String OPERATOR_PREFIX; static final String OPERATOR_CLASSNAME; static final String OPERATOR_TEMPLATE; static final String TEMPLATE_idRegExp; static final String TEMPLATE_appNameRegExp; static final String TEMPLATE_classNameRegExp; static final String CLASS; static final String KEY_SEPARATOR; static final String KEY_SEPARATOR_SPLIT_REGEX; static final String KEY_APPLICATION_NAME; static final String KEY_GATEWAY_CONNECT_ADDRESS; static final String KEY_GATEWAY_USE_SSL; static final String KEY_GATEWAY_USER_NAME; static final String KEY_GATEWAY_PASSWORD; final Configuration conf; }
@Test public void testLoadFromConfigXml() { Configuration conf = new Configuration(false); conf.addResource(StramClientUtils.DT_SITE_XML_FILE); LogicalPlanConfiguration builder = new LogicalPlanConfiguration(conf); LogicalPlan dag = new LogicalPlan(); builder.populateDAG(dag); dag.validate(); assertEquals("number of operator confs", 6, dag.getAllOperators().size()); OperatorMeta operator1 = assertNode(dag, "operator1"); OperatorMeta operator2 = assertNode(dag, "operator2"); OperatorMeta operator3 = assertNode(dag, "operator3"); OperatorMeta operator4 = assertNode(dag, "operator4"); assertNotNull("operatorConf for root", operator1); assertEquals("operatorId set", "operator1", operator1.getName()); assertEquals(operator1.getOperator().getClass(), TestGeneratorInputOperator.class); TestGeneratorInputOperator GenericTestNode = (TestGeneratorInputOperator)operator1.getOperator(); assertEquals("myStringPropertyValue", GenericTestNode.getMyStringProperty()); assertEquals("operator1 inputs", 0, operator1.getInputStreams().size()); assertEquals("operator1 outputs", 1, operator1.getOutputStreams().size()); StreamMeta n1n2 = operator2.getInputStreams().get(operator2.getMeta(((GenericTestOperator)operator2.getOperator()).inport1)); assertNotNull("n1n2", n1n2); assertEquals("rootNode out is operator2 in", n1n2, operator1.getOutputStreams().get(operator1.getMeta(((TestGeneratorInputOperator)operator1.getOperator()).outport))); assertEquals("n1n2 source", operator1, n1n2.getSource().getOperatorMeta()); Assert.assertEquals("n1n2 targets", 1, n1n2.getSinks().size()); Assert.assertEquals("n1n2 target", operator2, n1n2.getSinks().iterator().next().getOperatorMeta()); assertEquals("stream name", "n1n2", n1n2.getName()); Assert.assertEquals("n1n2 not inline (default)", null, n1n2.getLocality()); assertEquals("operator 2 number of outputs", 1, operator2.getOutputStreams().size()); StreamMeta fromNode2 = operator2.getOutputStreams().values().iterator().next(); Set<OperatorMeta> targetNodes = Sets.newHashSet(); for (LogicalPlan.InputPortMeta ip : fromNode2.getSinks()) { targetNodes.add(ip.getOperatorMeta()); } Assert.assertEquals("outputs " + fromNode2, Sets.newHashSet(operator3, operator4), targetNodes); OperatorMeta operator6 = assertNode(dag, "operator6"); List<OperatorMeta> rootNodes = dag.getRootOperators(); assertEquals("number root operators", 2, rootNodes.size()); assertTrue("root operator2", rootNodes.contains(operator1)); assertTrue("root operator6", rootNodes.contains(operator6)); for (OperatorMeta n : rootNodes) { printTopology(n, dag, 0); } }