Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()

[eolivelli]

Is your enhancement request related to a problem? Please describe.
In Pulsar IO we have Record#getKey that returns an Optional, this way you cannot support non String keys.

You can write Non string keys with:

• with KeyValue
• With MessageBuilder#keyBytes

Describe the solution you'd like
I would like to see a Record#getKeyObject() (or other name) that returns the key as Object:

• String if it is a string
• byte[] if we have a base64encoded byte[]
• Object in case of KeyValue payload (especially with SEPARATED encoding)

Describe alternatives you've considered
Using Record#getMessage#getKeyBytes() but it is not required for a Record to carry the original message

Additional context
This feature is needed in order to implement more advanced Sinks, in particular to be able to map the record key from Pulsar to Kafka and also in order to leverage Records that are using a GenericRecord as Key (see KeyValue with SEPARATED encoding)

[eolivelli]

cc @dlg99 @vroyer

[vroyer]

As the message value, the message key could be Json, Avro, Protobuf or whatever. So we could have a way to store byte[] returned by Schema.encode() as a base64.

[eolivelli]

Here I am talking more about the Sink part, that is when we read from Pulsar.
But yes, my idea is that we are able to ready the "logical key", whatever it is.

• raw byte[]
• String
• KeyValue - SEPARATED
• KeyValue - INLINE

I believe that the optimal situation it to see a "Record" as a pair of Object,Object
record.getValue() -> logical "value"
record.getKeyObject() -> logical "key"

maybe a better name is getLogicalKey ?

[vroyer]

May you coud have in Record:
default Optional<byte[]> getKeyBytes() {
return getKey().map(s -> Base64.getDecoder().decode(s));
}

And if the schema is a KeyValueSchema<Object,Object>, you should get the actual ket with getKey() properly decoded with the ObjectSchema of the key.

[eolivelli]

That was my first idea.
We can add that as well.

But in Pulsar we are used to automatic decoding of the value, so probably it is good to add getLogicalKey as well, otherwise the Sink code will have to deal explicitly with KeyValueSchema and I believe that it is some kind of implementation detail that is better to hide.

In my mind I have that only the KeyValue class is usable by the Sink code, like GenericRecord

[vroyer]

I think Record, Sink and Source should be concret class of KeyValue<Object,Object> , because we want generic connectors working whatever is the schema or encoding. So we would have Record like this:

   /**
     * Return the SEPARATED key if the message has one associated (stored in metdata properties).
     */
    default Optional<String> getKey() {
        return Optional.empty();
    }

    default Optional<byte[]> getKeyBytes() {
       return getKey().map(s -> Base64.getDecoder().decode(s));
    }

    default KeyValueSchema<Object,Object> getSchema() {
        return null;
    }

    /**
     * Retrieves the actual data of the record.
     *
     * @return The record data
     */
    KeyValue<Object, Object> getValue();

    default public Optional<Object> getRecordKey() {
        if (getSchema() != null) {
            switch(getSchema().getKeyValueEncodingType()) {
                case SEPARATED:
                    Schema<Object> objectSchema = getSchema().getKeySchema();
                    return (objectSchema != null)
                            ? Optional.of(objectSchema.decode(getKeyBytes().get()))
                            : Optional.of(getKey().get());
                case INLINE:
                    return Optional.of(getValue().getKey());
            }
        }
        return Optional.of(getValue().getKey());
    }

    default public Object getRecordValue() {
        return getValue().getValue();
    }

The sink connectors would use getRecordKey() and getRecordValue() to get the concret key and value object, and they can query the schema to process it properly, and whatever the key is INLINE or SEPARATED, like with Kafka connect :-)

[eolivelli]

@vroyer
that's something I would like to achieve but we have to take into account compatibility with the past

with #9896 we will open the way to Sink , and in that case Record.getValue() will contain your KeyValue object if you are coming from a KeyValue encoded message.

we could also create a new interface KeyValueRecord with your code and add some
Record#toKeyValueRecord() converter that maps the Record to the new API

[vroyer]

Why pulsar functions depends on pulsar-io Record ? Record is not only used by pulsar-io connectors ?

[eolivelli]

Pulsar-IO Sinks and Sources are special instances of Functions :-)

[eolivelli]

Partial support for KeyValue records with separate schema for Key and Value is provided by

pulsar/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/KVRecord.java

Line 31 in 3b2c852

public interface KVRecord<K, V> extends Record {

but it looks useful only to Sources

it is used for Kafka Connect Source, that encodes directly the key and the value to byte[]

pulsar/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java

Line 76 in 86e39a2

private class KafkaSourceRecord extends AbstractKafkaSourceRecord<KeyValue<byte[], byte[]>> implements KVRecord<byte[], byte[]> {

[eolivelli]

It was implemented by @tuteng in #6034