XMLParser extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); outSchema = config.getOutputSchema(collector); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (!this.config.containsMacro(Config.INPUT) && inputSchema != null && inputSchema.getField(this.config.inputField) == null) { collector.addFailure( String.format("Input field '%s' must exist in the input schema.", this.config.inputField), null) .withConfigProperty(Config.INPUT); } validateXpathAndSchema(collector); pipelineConfigurer.getStageConfigurer().setOutputSchema(outSchema); } XMLParser(Config config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter);  }
@Test public void testInvalidConfig() throws Exception { XMLParser.Config config = new XMLParser.Config("body", "UTF-8", "category: "year:/bookstore/book[price>35.00]/year,price:/bookstore/book[price>35.00]/price,subcategory: "category:,title:string,price:double,year:int,subcategory:string", "Exit on error"); MockPipelineConfigurer configurer = new MockPipelineConfigurer(INPUT); try { new XMLParser(config).configurePipeline(configurer); Assert.fail(); } catch (ValidationException e) { Assert.assertEquals(1, e.getFailures().size()); Assert.assertEquals(1, e.getFailures().get(0).getCauses().size()); Cause expectedCause = new Cause(); expectedCause.addAttribute("stage", "mockstage"); expectedCause.addAttribute(CauseAttributes.STAGE_CONFIG, XMLParser.Config.FIELD_TYPE_MAPPING); expectedCause.addAttribute(CauseAttributes.CONFIG_ELEMENT, "category:"); Assert.assertEquals(expectedCause, e.getFailures().get(0).getCauses().get(0)); } }
@Test public void testInputFieldNotInSchema() throws Exception { Schema schema = Schema.recordOf("record", Schema.Field.of("title", Schema.nullableOf(Schema.of(Schema.Type.STRING))), Schema.Field.of("author", Schema.nullableOf(Schema.of(Schema.Type.STRING))), Schema.Field.of("year", Schema.nullableOf(Schema.of(Schema.Type.STRING)))); XMLParser.Config config = new XMLParser.Config( "x", "UTF-8", "title:/book/title,author:/book/author,year:/book/year", "title:string,author:string,year:string", "Write to error dataset"); MockPipelineConfigurer configurer = new MockPipelineConfigurer(INPUT); new XMLParser(config).configurePipeline(configurer); FailureCollector collector = configurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(1, collector.getValidationFailures().get(0).getCauses().size()); Cause expectedCause = new Cause(); expectedCause.addAttribute(CauseAttributes.STAGE_CONFIG, XMLParser.Config.INPUT); Assert.assertEquals(expectedCause, collector.getValidationFailures().get(0).getCauses().get(0)); }