LogicalPlanConfiguration { public static GenericOperator setOperatorProperties(GenericOperator operator, Map<String, String> properties) { try { BeanUtils.populate(operator, properties); return operator; } catch (IllegalAccessException | InvocationTargetException e) { throw new IllegalArgumentException("Error setting operator properties", e); } } 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 testSetOperatorProperties() { Configuration conf = new Configuration(false); conf.set(StreamingApplication.APEX_PREFIX + "operator.o1.prop.myStringProperty", "myStringPropertyValue"); conf.set(StreamingApplication.APEX_PREFIX + "operator.o2.prop.stringArrayField", "a,b,c"); conf.set(StreamingApplication.APEX_PREFIX + "operator.o2.prop.mapProperty.key1", "key1Val"); conf.set(StreamingApplication.APEX_PREFIX + "operator.o2.prop.mapProperty(key1.dot)", "key1dotVal"); conf.set(StreamingApplication.APEX_PREFIX + "operator.o2.prop.mapProperty(key2.dot)", "key2dotVal"); LogicalPlan dag = new LogicalPlan(); GenericTestOperator o1 = dag.addOperator("o1", new GenericTestOperator()); ValidationTestOperator o2 = dag.addOperator("o2", new ValidationTestOperator()); LogicalPlanConfiguration pb = new LogicalPlanConfiguration(conf); pb.setOperatorProperties(dag, "testSetOperatorProperties"); Assert.assertEquals("o1.myStringProperty", "myStringPropertyValue", o1.getMyStringProperty()); Assert.assertArrayEquals("o2.stringArrayField", new String[] {"a", "b", "c"}, o2.getStringArrayField()); Assert.assertEquals("o2.mapProperty.key1", "key1Val", o2.getMapProperty().get("key1")); Assert.assertEquals("o2.mapProperty(key1.dot)", "key1dotVal", o2.getMapProperty().get("key1.dot")); Assert.assertEquals("o2.mapProperty(key2.dot)", "key2dotVal", o2.getMapProperty().get("key2.dot")); }