### Question:
SystemMetadataWriterStage extends AbstractStage<ApplicationWithPrograms> { @Override public void process(ApplicationWithPrograms input) { creationTime = String.valueOf(System.currentTimeMillis()); ApplicationId appId = input.getApplicationId(); ApplicationSpecification appSpec = input.getSpecification(); List<MetadataMutation> mutations = new ArrayList<>(); mutations.add( new AppSystemMetadataWriter(metadataServiceClient, appId, appSpec, creationTime).getMetadataMutation()); collectProgramSystemMetadata(appId, ProgramType.MAPREDUCE, appSpec.getMapReduce().values(), mutations); collectProgramSystemMetadata(appId, ProgramType.SERVICE, appSpec.getServices().values(), mutations); collectProgramSystemMetadata(appId, ProgramType.SPARK, appSpec.getSpark().values(), mutations); collectProgramSystemMetadata(appId, ProgramType.WORKER, appSpec.getWorkers().values(), mutations); collectProgramSystemMetadata(appId, ProgramType.WORKFLOW, appSpec.getWorkflows().values(), mutations); metadataServiceClient.batch(mutations); emit(input); } SystemMetadataWriterStage(MetadataServiceClient metadataServiceClient); @Override void process(ApplicationWithPrograms input);  }### Answer:
@Test public void testWorkflowTags() throws Exception { String appName = WorkflowAppWithFork.class.getSimpleName(); ApplicationId appId = NamespaceId.DEFAULT.app(appName); String workflowName = WorkflowAppWithFork.WorkflowWithFork.class.getSimpleName(); ArtifactId artifactId = NamespaceId.DEFAULT.artifact(appId.getApplication(), "1.0"); ApplicationWithPrograms appWithPrograms = createAppWithWorkflow(artifactId, appId, workflowName); WorkflowSpecification workflowSpec = appWithPrograms.getSpecification().getWorkflows().get(workflowName); SystemMetadataWriterStage systemMetadataWriterStage = new SystemMetadataWriterStage(metadataServiceClient); StageContext stageContext = new StageContext(Object.class); systemMetadataWriterStage.process(stageContext); systemMetadataWriterStage.process(appWithPrograms); Assert.assertEquals(false, metadataStorage.read(new Read(appId.workflow(workflowName).toMetadataEntity(), MetadataScope.SYSTEM, MetadataKind.TAG)).isEmpty()); Set<String> workflowSystemTags = metadataStorage .read(new Read(appId.workflow(workflowName).toMetadataEntity())).getTags(MetadataScope.SYSTEM); Sets.SetView<String> intersection = Sets.intersection(workflowSystemTags, getWorkflowForkNodes(workflowSpec)); Assert.assertTrue("Workflows should not be tagged with fork node names, but found the following fork nodes " + "in the workflow's system tags: " + intersection, intersection.isEmpty()); Assert.assertEquals(false, metadataStorage.read(new Read(appId.toMetadataEntity(), MetadataScope.SYSTEM, MetadataKind.PROPERTY)).isEmpty()); Map<String, String> metadataProperties = metadataStorage .read(new Read(appId.toMetadataEntity())).getProperties(MetadataScope.SYSTEM); Assert.assertEquals(WorkflowAppWithFork.SCHED_NAME + ":testDescription", metadataProperties.get("schedule:" + WorkflowAppWithFork.SCHED_NAME)); }