JSONFormatter extends Transform<StructuredRecord, StructuredRecord> { public JSONFormatter(Config config) { this.config = config; } JSONFormatter(Config config); @Override void initialize(TransformContext context); @Override void prepareRun(StageSubmitterContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter);  }
@Test public void testJSONFormatter() throws Exception { JSONFormatter.Config config = new JSONFormatter.Config(OUTPUT1.toString()); Transform<StructuredRecord, StructuredRecord> transform = new JSONFormatter(config); TransformContext context = new MockTransformContext(); transform.initialize(context); MockEmitter<StructuredRecord> emitter = new MockEmitter<>(); transform.transform(StructuredRecord.builder(INPUT1) .set("a", "1") .set("b", "2") .set("c", "3") .set("d", "4") .set("e", "5").build(), emitter); Assert.assertEquals("{\"a\":\"1\",\"b\":\"2\",\"c\":\"3\",\"d\":\"4\",\"e\":\"5\"}", emitter.getEmitted().get(0).get("body")); }