protected abstract PTransform<PCollection<KafkaRecord<byte[], byte[]>>, PCollection<Row>>
      getPTransformForInput();
@Test
  public void testRecorderDecoder() throws Exception {
    BeamKafkaTable kafkaTable = getBeamKafkaTable();

    PCollection<Row> result =
        pipeline
            .apply(Create.of(generateEncodedPayload(1), generateEncodedPayload(2)))
            .apply(MapElements.via(new BytesToRecord()))
            .setCoder(KafkaRecordCoder.of(ByteArrayCoder.of(), ByteArrayCoder.of()))
            .apply(kafkaTable.getPTransformForInput());

    PAssert.that(result).containsInAnyOrder(generateRow(1), generateRow(2));
    pipeline.run();
  }