You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/03/12 13:22:11 UTC
[GitHub] [pulsar] eolivelli opened a new issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
eolivelli opened a new issue #9899:
URL: https://github.com/apache/pulsar/issues/9899
**Is your enhancement request related to a problem? Please describe.**
In Pulsar IO we have Record#getKey that returns an Optional<String>, 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)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-800062091
It was implemented by @tuteng in #6034
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-1058892573
The issue had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-1084172073
now you can use Sink<GenericObject> and in case of KeyValue you will get the KeyValue object in record.getValue().getNativeObject()
closing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-800061449
Partial support for KeyValue records with separate schema for Key and Value is provided by https://github.com/apache/pulsar/blob/3b2c8526a96fc90e948dbeb510e8400a628c749a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/KVRecord.java#L31
but it looks useful only to Sources
it is used for Kafka Connect Source, that encodes directly the key and the value to byte[]
https://github.com/apache/pulsar/blob/86e39a25e961d148a485be4a7c9fc186d5b8f8bc/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java#L76
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797575292
@vroyer
that's something I would like to achieve but we have to take into account compatibility with the past
with https://github.com/apache/pulsar/pull/9896 we will open the way to Sink<Object> , 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
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797524691
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
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] vroyer commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
vroyer commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797570247
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797581213
Pulsar-IO Sinks and Sources are special instances of Functions :-)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] vroyer commented on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
vroyer commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797504582
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797507997
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
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli closed issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
eolivelli closed issue #9899:
URL: https://github.com/apache/pulsar/issues/9899
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli edited a comment on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
eolivelli edited a comment on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797486724
cc @dlg99 @vroyer
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
eolivelli commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797486724
cc @dlg99
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] vroyer commented on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
vroyer commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797513900
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] vroyer commented on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
vroyer commented on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797575486
Why pulsar functions depends on pulsar-io Record ? Record is not only used by pulsar-io connectors ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli edited a comment on issue #9899: Support non String keys in Pulsar IO: Record#getKeyObject()
Posted by GitBox <gi...@apache.org>.
eolivelli edited a comment on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797507997
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` ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] vroyer edited a comment on issue #9899: Support non String keys in Pulsar IO: Record#getLogicalKey() and Record#getKeyBytes()
Posted by GitBox <gi...@apache.org>.
vroyer edited a comment on issue #9899:
URL: https://github.com/apache/pulsar/issues/9899#issuecomment-797570247
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 :-)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org