Support Pulsar Message as function input

[jerrypeng]

Motivation

Support Message object as function input so that users can get all of the attributes of the Message.

Implementing this for windowed function is a more complicated and will discuss/implement after this

If everyone is ok with this approach then I will add tests

Discussions about this was done in:

#4127

[jerrypeng]

@ConcurrencyPractitioner please also take a look

[sijie]

Can you add an integration test for this?

[jerrypeng]

@sijie as mentioned in the description of the PR, once everyone is on board with this approach I will add the tests for this

[jerrypeng]

@sijie I have added the tests

[srkukarni]

whats the plan for adding this support to window function?

[jerrypeng]

we will probably have to change the interface from

public interface WindowFunction<I, O> {
    O process(Collection<Record<I>> input, WindowContext context) throws Exception;
}

to

public interface WindowFunction<I, O> {
    O process(Collection<I> input, WindowContext context) throws Exception;
}

And support "I" as Record for BC and support "I" as Message.

[ConcurrencyPractitioner]

Sure, I have no problem with this. Thanks for picking this up. :)

[jerrypeng]

rerun cpp tests

[jerrypeng]

@srkukarni @sijie can you look at this again? Thanks

[jerrypeng]

Alternatively, instead of allowing user's to be able to have Message as input to a function, a simpler approach of we would be just to add this API to Context:

 Message getInternalMessage();

and the Impl would be:

Message getInternalMessage() {
return ((PulsarRecord) this.record).getMessage);
}

There also isn't any good way (without breaking BC) for the WindowFunction API to support Message as a input since the WindowFunction API is

public interface WindowFunction<I, O> {
    /**
     * Process the input.
     *
     * @return the output
     */
    O process(Collection<Record<I>> input, WindowContext context) throws Exception;
}

I don't think it makes much sense to do

    O process(Collection<Record<Message<I>>> input, WindowContext context) throws Exception;

We would have to change the interface (BC breaking) to:

public interface WindowFunction<I, O> {
    /**
     * Process the input.
     *
     * @return the output
     */
    O process(Collection<I> input, WindowContext context) throws Exception;
}

To support Message as a input

Alternatively, we can change the API for WindowFunction to be something like

    O process(Window<I> input, WindowContext context) throws Exception;

and you can do

input.getAsMessages()

or 

input.getAsRecords()

or in the WindowContext we implement a method like

context.getWindowAsMessages()

[jerrypeng]

The simplest and non-BC-breaking approach I can think of for Functions and WindowFunctions:

just to add these getter methods to Context and WindowContext:

Message getInternalMessage();

and

List<Message> getWindowAsMessages()

[merlimat]

@jerrypeng What about the message type? I believe Context is not generic at this point. Would you get a Message<?> then?

[jerrypeng]

@merlimat that is correct which is a con for that approach.

[jerrypeng]

@sijie @srkukarni @merlimat what we can do is this:
#4341

simpler approach that solves all our problems

[srkukarni]

+1 for #4341