chore: support PROTOBUF for PRINT TOPIC

[agavra]

fixes #4516

Description

Adds support for PRINT TOPIC for protobuf topics

Testing done

Unit testing and manual:

ksql> PRINT print_proto FROM BEGINNING;
Key format: UNDEFINED
Value format: PROTOBUF
rowtime: 2/19/20 9:40:02 AM PST, key: <null>, value: COL1: "a" COL2: 2 COL3: "c"
^CTopic printing ceased

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[hjafarpour]

Key format: UNDEFINED
Value format: PROTOBUF

@agavra usually Key and Value have the same format when using schema registry. What does UNDEFINED mean in this context? Do we show UNDEFINED if the key format is PROTOBUF?

[agavra]

@hjafarpour - non-KAFKA keys are not supported at all yet. It is UNDEFINED since all of the keys so far were null. If it was long or int or string it would've shown that value.

[hjafarpour]

Thanks @agavra , can't wait to have AVRO/JSON/PROTOBUF key support! ;)

[purplefox]

LGTM

[big-andy-coates]

Thanks @agavra.

LGTM, except I don't think we need to complicate the RecordFormatter code by adding a format method.

[big-andy-coates]

nit: generally avoid logging and throwing - it's an anti-pattern. Why isn't it sufficient just to throw here? Does the calling code not log it?

[agavra]

the calling code doesn't log it - it gets returned to the user in the CLI as "could not serialize row" but it doesn't give the full trace. This has bit me several times so I'd rather keep it here as at some point we should log it (we can debate where, I don't care that much)

[big-andy-coates]

This call to format seems completely superfluous. This method now calls deserialize on the deserializer and then immediately passes the result back to format on the same deserializer. Why not just have deserialize return the formatted result?

In the case of the new protobuf format, all we need to do is pass a factory method to the constructor of the PROTOBUF enum that wraps the KafkaProtobufDeserializer instance in one that applies the format, in a similar way to newJsonDeserializer.

   private static Deserializer<?> newProtobufDeserializer(final SchemaRegistryClient srClient) {

    final Printer printer = TextFormat.printer();
    final KafkaProtobufDeserializer<?> inner = new KafkaProtobufDeserializer<>(srClient);
    
    return (Deserializer<Object>) (topic, data) -> {
      final Message msg = inner.deserialize(topic, data);
      if (msg == null) {
        return null;
      }
      
      return printer.shortDebugString(msg);
    };
  }

...

 enum Format {
    AVRO(0, KafkaAvroDeserializer::new),
    PROTOBUF(0, RecordFormatter::newAvroDeserializer),
    JSON(RecordFormatter::newJsonDeserializer),
 ...
 } 

This avoids the need to complicate the code with another function call on all deserializers...