Encoder extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.encode, collector); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { for (Schema.Field field : inputSchema.getFields()) { if (encodeMap.containsKey(field.getName())) { Schema.Type type = field.getSchema().getType(); if (!type.equals(Schema.Type.BYTES) && !type.equals(Schema.Type.STRING)) { collector.addFailure(String.format("Field '%s' has invalid input type '%s'.", field.getName(), type.toString().toLowerCase()), "Input schema field must be of type string or bytes.") .withConfigElement(Config.ENCODE, String.format("%s:%s", field.getName(), type.toString().toLowerCase())) .withInputSchemaField(field.getName()); } } } } try { Schema outputSchema = Schema.parseJson(config.schema); pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } catch (IOException e) { collector.addFailure("Invalid schema format.", null).withConfigProperty(Config.SCHEMA); } } Encoder(Config config); @Override void prepareRun(StageSubmitterContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord in, Emitter<StructuredRecord> emitter);  }
@Test public void testSchemaValidation() throws Exception { Transform<StructuredRecord, StructuredRecord> transform = new Encoder(new Encoder.Config("a:BASE32", OUTPUTSTR.toString())); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(INPUT); transform.configurePipeline(mockPipelineConfigurer); Assert.assertEquals(OUTPUTSTR, mockPipelineConfigurer.getOutputSchema()); FailureCollector collector = mockPipelineConfigurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(0, collector.getValidationFailures().size()); }
@Test public void testSchemaValidationWithInvalidInputSchema() throws Exception { Transform<StructuredRecord, StructuredRecord> transform = new Encoder(new Encoder.Config("a:BASE32", OUTPUTSTR.toString())); final Schema invalidInput = Schema.recordOf("input", Schema.Field.of("a", Schema.of(Schema.Type.INT)), Schema.Field.of("b", Schema.of(Schema.Type.STRING)), Schema.Field.of("c", Schema.of(Schema.Type.STRING))); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(invalidInput); transform.configurePipeline(mockPipelineConfigurer); FailureCollector collector = mockPipelineConfigurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(2, collector.getValidationFailures().get(0).getCauses().size()); }