LogicalPlanConfiguration { public LogicalPlan createFromJson(JSONObject json, String appName) throws Exception { LogicalPlanConfiguration tb = new LogicalPlanConfiguration(new Configuration(false)); tb.addFromJson(json, conf); return populateDAGAndValidate(tb, appName); } 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 testLoadFromJson() throws Exception { String resourcePath = "/testTopology.json"; InputStream is = this.getClass().getResourceAsStream(resourcePath); if (is == null) { fail("Could not load " + resourcePath); } StringWriter writer = new StringWriter(); IOUtils.copy(is, writer); JSONObject json = new JSONObject(writer.toString()); Configuration conf = new Configuration(false); conf.set(StreamingApplication.APEX_PREFIX + "operator.operator3.prop.myStringProperty", "o3StringFromConf"); LogicalPlanConfiguration planConf = new LogicalPlanConfiguration(conf); LogicalPlan dag = planConf.createFromJson(json, "testLoadFromJson"); dag.validate(); assertEquals("DAG attribute CONTAINER_JVM_OPTIONS ", dag.getAttributes().get(DAGContext.CONTAINER_JVM_OPTIONS), "-Xmx16m"); Map<Class<?>, Class<? extends StringCodec<?>>> stringCodecsMap = Maps.newHashMap(); stringCodecsMap.put(Integer.class, Integer2String.class); assertEquals("DAG attribute STRING_CODECS ", stringCodecsMap, dag.getAttributes().get(DAGContext.STRING_CODECS)); assertEquals("DAG attribute CONTAINER_OPTS_CONFIGURATOR ", BasicContainerOptConfigurator.class, dag.getAttributes().get(DAGContext.CONTAINER_OPTS_CONFIGURATOR).getClass()); assertEquals("number of operator confs", 5, dag.getAllOperators().size()); assertEquals("number of root operators", 1, dag.getRootOperators().size()); StreamMeta s1 = dag.getStream("n1n2"); assertNotNull(s1); assertTrue("n1n2 inline", DAG.Locality.CONTAINER_LOCAL == s1.getLocality()); OperatorMeta input = dag.getOperatorMeta("inputOperator"); TestStatsListener tsl = new TestStatsListener(); tsl.setIntProp(222); List<StatsListener> sll = Lists.<StatsListener>newArrayList(tsl); assertEquals("inputOperator STATS_LISTENERS attribute ", sll, input.getAttributes().get(OperatorContext.STATS_LISTENERS)); for (OutputPortMeta opm : input.getOutputStreams().keySet()) { assertTrue("output port of input Operator attribute is JsonStreamCodec ", opm.getAttributes().get(PortContext.STREAM_CODEC) instanceof JsonStreamCodec<?>); } OperatorMeta operator3 = dag.getOperatorMeta("operator3"); assertEquals("operator3.classname", GenericTestOperator.class, operator3.getOperator().getClass()); GenericTestOperator doperator3 = (GenericTestOperator)operator3.getOperator(); assertEquals("myStringProperty " + doperator3, "o3StringFromConf", doperator3.getMyStringProperty()); assertFalse("booleanProperty " + doperator3, doperator3.booleanProperty); OperatorMeta operator4 = dag.getOperatorMeta("operator4"); GenericTestOperator doperator4 = (GenericTestOperator)operator4.getOperator(); assertEquals("myStringProperty " + doperator4, "overrideOperator4", doperator4.getMyStringProperty()); assertEquals("setterOnlyOperator4 " + doperator4, "setterOnlyOperator4", doperator4.propertySetterOnly); assertTrue("booleanProperty " + doperator4, doperator4.booleanProperty); StreamMeta input1 = dag.getStream("inputStream"); assertNotNull(input1); OperatorMeta inputOperator = dag.getOperatorMeta("inputOperator"); Assert.assertEquals("input1 source", inputOperator, input1.getSource().getOperatorMeta()); Set<OperatorMeta> targetNodes = Sets.newHashSet(); for (LogicalPlan.InputPortMeta targetPort : input1.getSinks()) { targetNodes.add(targetPort.getOperatorMeta()); } Assert.assertEquals("operator attribute " + inputOperator, 64, (int)inputOperator.getValue(OperatorContext.MEMORY_MB)); Assert.assertEquals("port attribute " + inputOperator, 8, (int)input1.getSource().getValue(PortContext.UNIFIER_LIMIT)); Assert.assertEquals("input1 target ", Sets.newHashSet(dag.getOperatorMeta("operator1"), operator3, operator4), targetNodes); }
@Test public void testTupleClassAttr() throws Exception { String resourcePath = "/schemaTestTopology.json"; InputStream is = this.getClass().getResourceAsStream(resourcePath); if (is == null) { fail("Could not load " + resourcePath); } StringWriter writer = new StringWriter(); IOUtils.copy(is, writer); JSONObject json = new JSONObject(writer.toString()); Configuration conf = new Configuration(false); LogicalPlanConfiguration planConf = new LogicalPlanConfiguration(conf); LogicalPlan dag = planConf.createFromJson(json, "testLoadFromJson"); dag.validate(); OperatorMeta operator1 = dag.getOperatorMeta("operator1"); assertEquals("operator1.classname", SchemaTestOperator.class, operator1.getOperator().getClass()); StreamMeta input1 = dag.getStream("inputStream"); assertNotNull(input1); for (LogicalPlan.InputPortMeta targetPort : input1.getSinks()) { Assert.assertEquals("tuple class name required", TestSchema.class, targetPort.getAttributes().get(PortContext.TUPLE_CLASS)); } }
@Test(expected = ValidationException.class) public void testTupleClassAttrValidation() throws Exception { String resourcePath = "/schemaTestTopology.json"; InputStream is = this.getClass().getResourceAsStream(resourcePath); if (is == null) { fail("Could not load " + resourcePath); } StringWriter writer = new StringWriter(); IOUtils.copy(is, writer); JSONObject json = new JSONObject(writer.toString()); json.getJSONArray("streams").getJSONObject(0).remove("schema"); Configuration conf = new Configuration(false); LogicalPlanConfiguration planConf = new LogicalPlanConfiguration(conf); LogicalPlan dag = planConf.createFromJson(json, "testLoadFromJson"); dag.validate(); }