@Test public void testSchemaInvalidSchema() throws Exception { JSONParser.Config config = new JSONParser.Config("body2", "", OUTPUT2.toString()); Transform<StructuredRecord, StructuredRecord> transform = new JSONParser(config); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(INPUT1); transform.configurePipeline(mockPipelineConfigurer); FailureCollector collector = mockPipelineConfigurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(1, collector.getValidationFailures().get(0).getCauses().size()); ValidationFailure.Cause expectedCause = new ValidationFailure.Cause(); expectedCause.addAttribute(CauseAttributes.STAGE_CONFIG, JSONParser.Config.FIELD); Assert.assertEquals(expectedCause, collector.getValidationFailures().get(0).getCauses().get(0)); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema outputSchema = Schema.parseJson(config.schema); pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); fields = outputSchema.getFields(); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON."); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null && inputSchema.getField(config.field) == null) { collector.addFailure(String.format("Field '%s' must be present in the input schema.", config.field), null) .withConfigProperty(Config.FIELD); } extractMappings(collector); }
JSONParser extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema outputSchema = Schema.parseJson(config.schema); pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); fields = outputSchema.getFields(); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON."); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null && inputSchema.getField(config.field) == null) { collector.addFailure(String.format("Field '%s' must be present in the input schema.", config.field), null) .withConfigProperty(Config.FIELD); } extractMappings(collector); } }
JSONParser extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema outputSchema = Schema.parseJson(config.schema); pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); fields = outputSchema.getFields(); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON."); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null && inputSchema.getField(config.field) == null) { collector.addFailure(String.format("Field '%s' must be present in the input schema.", config.field), null) .withConfigProperty(Config.FIELD); } extractMappings(collector); } JSONParser(Config config); }
JSONParser extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema outputSchema = Schema.parseJson(config.schema); pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); fields = outputSchema.getFields(); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON."); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null && inputSchema.getField(config.field) == null) { collector.addFailure(String.format("Field '%s' must be present in the input schema.", config.field), null) .withConfigProperty(Config.FIELD); } extractMappings(collector); } JSONParser(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); }
JSONParser extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema outputSchema = Schema.parseJson(config.schema); pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); fields = outputSchema.getFields(); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON."); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null && inputSchema.getField(config.field) == null) { collector.addFailure(String.format("Field '%s' must be present in the input schema.", config.field), null) .withConfigProperty(Config.FIELD); } extractMappings(collector); } JSONParser(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);  }