PGETaskInstance implements WorkflowTaskInstance { protected PgeMetadata createPgeMetadata(Metadata dynMetadata, WorkflowTaskConfiguration config) { logger.info("Converting workflow configuration to static metadata"); logger.debug("PGE Metadata Config: {}", config.getProperties().entrySet()); logger.debug("PGE Metadata Dynamic Metadata: {}", dynMetadata.getMap()); Metadata staticMetadata = new Metadata(); for (Object objKey : config.getProperties().keySet()) { String key = (String) objKey; PgeTaskMetKeys metKey = PgeTaskMetKeys.getByName(key); if (metKey != null && metKey.isVector()) { List<String> values = Lists.newArrayList( Splitter.on(",").trimResults() .omitEmptyStrings() .split(config.getProperty(key))); logger.debug("Adding static metadata: key = [{}] value = {}", key, values); staticMetadata.addMetadata(key, values); } else { String value = config.getProperty(key); logger.debug("Adding static metadata: key = [{}] value = {}", key, value); staticMetadata.addMetadata(key, value); } } logger.info("Loading workflow context metadata..."); for (String key : dynMetadata.getAllKeys()) { logger.debug("Adding dynamic metadata: key = [{}] value = {}", key, dynMetadata.getAllMetadata(key)); } return new PgeMetadata(staticMetadata, dynMetadata); } protected  PGETaskInstance(); @Override void run(Metadata metadata, WorkflowTaskConfiguration config); String getWorkflowInstId(); void setWorkflowInstId(String workflowInstId); void setWmClient(WorkflowManagerClient wmClient); @Override void finalize();  }
@Test public void testCreatePgeMetadata() throws Exception { final String PGE_NAME = "PGE_Test"; final String PGE_REQUIRED_METADATA = "Filename, FileLocation "; final String PROP_ADDERS = "some.prop.adder.classpath,some.other.classpath"; PGETaskInstance pgeTask = createTestInstance(); Metadata dynMet = new Metadata(); WorkflowTaskConfiguration config = new WorkflowTaskConfiguration(); config.addConfigProperty(NAME.getName(), PGE_NAME); config.addConfigProperty(REQUIRED_METADATA.getName(), PGE_REQUIRED_METADATA); config.addConfigProperty(PROPERTY_ADDERS.getName(), PROP_ADDERS); PgeMetadata pgeMet = pgeTask.createPgeMetadata(dynMet, config); assertEquals(1, pgeMet.getAllMetadata(NAME).size()); assertEquals(PGE_NAME, pgeMet.getAllMetadata(NAME).get(0)); assertEquals(2, pgeMet.getAllMetadata(REQUIRED_METADATA).size()); assertTrue(pgeMet.getAllMetadata(REQUIRED_METADATA).contains("Filename")); assertTrue(pgeMet.getAllMetadata(REQUIRED_METADATA).contains( "FileLocation")); assertEquals(2, pgeMet.getAllMetadata(PROPERTY_ADDERS).size()); assertTrue(pgeMet.getAllMetadata(PROPERTY_ADDERS).contains( "some.prop.adder.classpath")); assertTrue(pgeMet.getAllMetadata(PROPERTY_ADDERS).contains( "some.other.classpath")); pgeTask = createTestInstance(); config = new WorkflowTaskConfiguration(); config.addConfigProperty(PgeTaskMetKeys.PROPERTY_ADDERS.getName(), "one.prop.adder.only"); pgeMet = pgeTask.createPgeMetadata(dynMet, config); assertEquals(1, pgeMet.getAllMetadata(PROPERTY_ADDERS).size()); assertEquals("one.prop.adder.only", pgeMet .getAllMetadata(PROPERTY_ADDERS).get(0)); }