How should I set up the message payload when calling a remote Python function from an embedded Java function

user_231762 0
Viewed 186

I'm using Flink StateFun 3.1.0, and want to call a remote Python function from an embedded Java function. In StateFun 2.2, I could create a ProtoBuf Any from an instance of my (generated via Protobuf) Java class via Any.pack(msg), and that worked. Now I get

Caused by: java.lang.ClassCastException: class com.google.protobuf.Any cannot be cast to class org.apache.flink.statefun.sdk.reqreply.generated.TypedValue (com.google.protobuf.Any and org.apache.flink.statefun.sdk.reqreply.generated.TypedValue are in unnamed module of loader 'app')

There are plenty of examples of calling between embedded Java functions, and remote Java functions, and remote Python functions, but I haven't found an example of calling remote Python from embedded Java.

In the remote Java examples, the function is called with a Context that has a send(Message) method, which I assume would work for calling a remote Python function as well, but my embedded Java function is passed a different type of Context that does not support this method.

flink-statefun
1 Answers

Background

Indeed in the 3.x release one of the explicit goals was to make remote functions easier and more ergonomic to use, you can read more about the motivation for that here.

To help achieving this goal we removed Protobuf from the SDK surface (we still use it internally) and we came up with the Message and types abstractions.

Since the embedded SDK is considered to be more of a power-user SDK, there you would still use Protobuf for communicating with remote functions. Although this is something we would like to simplify as well but there are some technical issues.

TypedValue

This is a Protobuf message that is used internally, whenever a remote function invokes another remote function.

message TypedValue {
    string typename = 1;
    // has_value is set to differentiate a zero length value bytes explicitly set,
    // or a non existing value.
    bool has_value = 2;
    bytes value = 3;
}

  • The typename field is a string of the form / that is defined by the user. for example: "com.kkrugler.types/MyEvent". StateFun will pass that string as-us, and treats that as an opaque type tag, to be interpreted by the user on the remote SDK side.

  • has_value in this context, needs to be always set to true. (it is considered an error to send a message without a payload.

  • values - is a serialised, opaque value of the type typename.

Embedded Functions

An embedded function that needs to invoke a remote function, needs invoke it with a TypedValue instance. The remote SDKs know how to convert a TypedValue into a Message class.

Drop in replacement for Any

Here is a useful method for wrapping a Protobuf message with a TypedValue:

 public static <M extends Message> TypedValue pack(M message) {
    return TypedValue.newBuilder()
        .setTypename("type.googleapis.com/" + message.getDescriptorForType().getFullName())
        .setHasValue(true)
        .setValue(message.toByteString())
        .build();
  }

Then, simply use it like that:

MyProtobufEvent myProtobufEvent = MyProtobufEvent.newBuilder() ... build();

context.send(.., pack(myProtobufEvent));

On the remote SDK side (assuming Python, since you've mentioned that):

from statefun import make_protobuf_type

MyProtobufEventType = make_protobuf_type(MyProtobufEvent)

...
myProtobufEvent = message.as_type(MyProtobufEventType)
user_44054700
Related