Decoder extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer(); FailureCollector collector = stageConfigurer.getFailureCollector(); parseConfiguration(config.decode, collector); validateInputSchema(stageConfigurer.getInputSchema(), collector); Schema outputSchema = config.getSchema(collector); if (outputSchema != null) { for (String fieldName : decodeMap.keySet()) { if (outputSchema.getField(fieldName) == null) { collector.addFailure(String.format("Field '%s' must be in output schema.", fieldName), null) .withConfigElement(Config.NAME_DECODE, fieldName + Config.SEPARATOR + decodeMap.get(fieldName)); continue; } Field field = outputSchema.getField(fieldName); Schema nonNullableSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema(); if (nonNullableSchema.getType() != Schema.Type.BYTES && nonNullableSchema.getType() != Schema.Type.STRING || nonNullableSchema.getLogicalType() != null) { collector.addFailure(String.format("Decode field '%s' is of invalid type '%s'.", field.getName(), nonNullableSchema.getDisplayName()), "Ensure the decode field is of type string or bytes.") .withOutputSchemaField(field.getName()) .withConfigElement(Config.NAME_DECODE, fieldName + Config.SEPARATOR + decodeMap.get(fieldName)); } } } stageConfigurer.setOutputSchema(outputSchema); } Decoder(Config config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord in, Emitter<StructuredRecord> emitter);  }
@Test public void testSchemaValidation() { Transform<StructuredRecord, StructuredRecord> decoder = new Decoder(new Decoder.Config("a:BASE64", OUTPUT.toString())); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(INPUT); decoder.configurePipeline(mockPipelineConfigurer); Assert.assertEquals(OUTPUT, mockPipelineConfigurer.getOutputSchema()); }
@Test public void testSchemaValidationWithInvalidInputSchema() { Transform<StructuredRecord, StructuredRecord> decoder = new Decoder(new Decoder.Config("a:BASE64", OUTPUT.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); decoder.configurePipeline(mockPipelineConfigurer); FailureCollector failureCollector = mockPipelineConfigurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(1, failureCollector.getValidationFailures().size()); }