anchor.diff:              context.output(toJson(context.element()));
           }
   
  -        // todo: we shouldnt be typed string/string so make it evolving
           private JsonObject toJson(final KafkaRecord<byte[], byte[]> element) {
               return builder.createObjectBuilder().add("key", new String(element.getKV().getKey()))
                       .add("value", new String(element.getKV().getValue())).build();
  
positive.code: }
  ----
  
  == Wrap an output
  
  Assume you want to wrap an output like this one (based on existing Beam ones):
  
  [source,java]
  ----
  
negative.code: }
  ----
  
  You can wrap this output exactly the same way you wrap an input, but using `@Processor` instead of:
  
  [source,java]
  ----
  @Processor(family = "myfamily", name = "myname")
  public class WrapWrite extends PTransform<PCollection<String>, PDone> {
    private PTransform<PCollection<String>, PDone> delegate;
  
    public WrapWrite(@Option("dataset") final WrapWriteDataSet dataset) {
      delegate = TheIO.write().withConfiguration(this.createConfigurationFrom(dataset));
    }
  
  