KafkaIO does not handle null keys

Kafka ConsumerRecord and ProducerRecord 'key' fields are explicitly allowed to be null. In addition, on the producer side, setting a null key is the way that the user indicates that they want a random partition for their message

Beam KafkaIO does not support null keys in byte[] mode (read side: https://github.com/apache/beam/blob/9e0997760cf3320f1a1d0c4342d3dff559a25775/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L727

write side: https://github.com/apache/beam/blob/9e0997760cf3320f1a1d0c4342d3dff559a25775/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoder.java#L58)

since it would defer to ByteArrayCoder which does not support null arrays.

BeamKafkaTable suffers the same issue https://github.com/apache/beam/blob/9e0997760cf3320f1a1d0c4342d3dff559a25775/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java#L144