@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)); } }
@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 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 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); }
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); }
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);  }