Function should support `Message<T>`

[sijie]

Is your feature request related to a problem? Please describe.

The message metadata is lost in source/function. People can not use these information for their computation.

Describe the solution you'd like

Similar as what was discussed in #4093, we should support using Message<T> as the input for a source or a function.

public class MessageFunction implements Function<Message<String>, String>> {
...
}

Describe alternatives you've considered

N/A

[jerrypeng]

@sijie aren't we using the Context object to expose these items?

More specifically:

public Record<?> getCurrentRecord() {
        return record;
    }

Which will return a PulsarRecord

https://github.com/apache/pulsar/blob/master/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarRecord.java

that contains the information about the input message

[sijie]

@jerrypeng yes. there is already a getCurrentRecord. However the Record interface doesn't catch up the Message interface, hence there are a few fields are not added. e.g. publish time, schema version and redelivery count. As I mentioned in the discussion in #4093 , the problem of having two interfaces is that it is really hard to make them consistent with each other.

There are a couple of options here:

1. we catch up the interface in Record api. but the problem still exists since anytime people make changes to Message api. The metadata will be first lost at Record api until someone add it.

2. we add getCurrentMessage to return the current message.

3. the proposal here.

I don't have any strong opinion with any approach here. However if we take 3), we might be able to support Message<T> in java native function.

[jerrypeng]

@sijie I am open for having inputs as Message<T>, my only concern is for users to only have one way to get the contextual information about the input whatever that might be and not duplicating functionality in different parts of the functions code base. It just gets confusing not only for the user of functions but also developers when you can get the same info via different methods.

Alternatively, we could also add a getActualMessage() method to PulsarRecord.

[sijie]

Alternatively, we could also add a getActualMessage() method to PulsarRecord.

I am fine with this approach.

[ConcurrencyPractitioner]

@jerrypeng If you hadn't started, could I try to take a hack at it?

[jerrypeng]

@ConcurrencyPractitioner sounds good! Thanks for taking this up!

[ConcurrencyPractitioner]

@jerrypeng Cool, thanks!
Just to clarify one thing, are we just adding this new method or is there something else that needs to be done?

[jerrypeng]

@ConcurrencyPractitioner just add this new method to PulsarRecord class for now.