[FLINK-19001] Add data stream api interoperability

[igalshilman]

Add DataStream API interoperability

This PR adds an extension the stateful function SDK that allows embedding stateful function applications into a data stream program.

This integration allows:

• defining DataStreams as StateFun ingresses
• binding one or more stateful functions
• binding remote RequestReply functions
• and finally obtaining StateFun egresses as DataStreams.

Here is a short example snippet of how to insert a StateFun pipeline into a regular DataStream pipeline. (full example)

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  DataStream<String> namesStream = ... ;  

  DataStream<RoutableMessage> names =
        namesStream
            .map(
                name ->
                    RoutableMessageBuilder.builder()
                        .withTargetAddress(GREET, name)
                        .withMessageBody(name)
                        .build());

  ...
  StatefulFunctionEgressStreams out =
        StatefulFunctionDataStreamBuilder.builder("example")
            .withDataStreamAsIngress(names)
            .withFunctionProvider(GREET, unused -> new MyFunction())
            .withRequestReplyRemoteFunction(
                requestReplyFunctionBuilder(
                        REMOTE_GREET, URI.create("http://localhost:5000/statefun"))
                    .withPersistedState("seen_count")
                    .withMaxRequestDuration(Duration.ofSeconds(15))
                    .withMaxNumBatchRequests(500))
            .withEgressId(GREETINGS)
            .withConfiguration(statefunConfig)
            .build(env);
  ...

• A RoutableMessage is the entry point to the StateFun pipeline.
• Then it is possible to register 1 or more DataStream<RoutableMessage> as ingresses.
• StateFun would deliver the payloads associated with the RoutableMessage to the appropriate stateful function instance.
• Egress ids has to be explicitly defined, and they can be collected as a DataStream from StatefulFunctionEgressStreams.

---

The following is a short description of the changes (not in the same commit order)

• Add a statefun-flink/statefun-flink-datastream for the new SDK.
• Added a new example under examples/
• Restructured the translation logic so that it can be used both from the data stream api and the regular statefun sdk.

[sjwiesman]

#docs

[igalshilman]

Would be followed on the docs sprint, thanks for the reminder.