ClassCastException occurs when Flink DataStream sends a message to a remote stateful function

[Tymur Yarosh <ti...@gmail.com>]

Hello, I’m trying to send a message from DataStream API to a remote Stateful Function (Java SDK). The DataStream job:


    public static final FunctionType DEVICE = new FunctionType("com.github.f1xman.era.anomalydetection.device", "DeviceFunction");

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StatefulFunctionsConfig statefunConfig = StatefulFunctionsConfig.fromEnvironment(env);
        statefunConfig.setFactoryType(MessageFactoryType.WITH_KRYO_PAYLOADS);

        DataStreamSource<String> names = env.addSource(new NamesSourceFunction());
        DataStream<RoutableMessage> namesIngress = names.map(name -> RoutableMessageBuilder.builder()
                .withTargetAddress(DEVICE, name)
                .withMessageBody(name)
                .build());
        StatefulFunctionDataStreamBuilder.builder("example")
                .withDataStreamAsIngress(namesIngress)
                .withRequestReplyRemoteFunction(
                        requestReplyFunctionBuilder(DEVICE, URI.create("http://localhost:8080/statefun"))
                )
                .withConfiguration(statefunConfig)
                .build(env);

        env.execute("Flink Streaming Java API Skeleton");
    }


When String value passed to .withMessageBody(...) the following exception occurred:


java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.flink.statefun.sdk.reqreply.generated.TypedValue
    at org.apache.flink.statefun.flink.core.reqreply.RequestReplyFunction.invoke(RequestReplyFunction.java:118)
    at org.apache.flink.statefun.flink.core.functions.StatefulFunction.receive(StatefulFunction.java:48)
    ... 25 more


Though, sending a String value to an embedded function works well. The workaround I've found is to wrap the value with TypedValue:

                .withMessageBody(TypedValue.newBuilder()
                        .setValue(ByteString.copyFrom(name, StandardCharsets.UTF_8))
                        .setHasValue(true)
                        .setTypename("example/Name")
                        .build()
                )

This approach requires the receiver function to unwrap the TypedValue and deserialize the ByteString. It looks too low-level for this kind of API. I believe this is the wrong usage of Stateful Function's SDK for Flink DataStream Integration. What is the correct way to implement Flink DataStream and remote Stateful Functions interoperability?

The job is inspired by the official examples https://nightlies.apache.org/flink/flink-statefun-docs-release-3.2/docs/sdk/flink-datastream/

Best,
Tymur Yarosh