Igal Shilman created FLINK-16123:
------------------------------------
Summary: Add routable Kafka connector
Key: FLINK-16123
URL: https://issues.apache.org/jira/browse/FLINK-16123
Project: Flink
Issue Type: Task
Components: Stateful Functions
Reporter: Igal Shilman
In some cases it is beneficial to associate a stateful function instance with a
key in a Kafka topic.
In that case, a simplified Kafka ingress definition can be introduced.
Consider the following example:
Imagine a Kafka topic named "signups" (1) where the keys are ut8 strings
representing user ids,
and the values are Protobuf messages of type (2)
com.user.foo.bar.greeter.SingupMessage.
We would like to have a stateful function of type(3)
{code:java}
FunctionType( com.user.foo.bar, SingupProcessor{code}
to be invoked for each incoming signup message.
The following spec definition:
{code:java}
- ingress:
meta:
type: org.apache.flink.statefun.sdk.kafka/routable-kafka-connector
id: com.user.foo.bar/greeter
spec:
properties:
- consumer.group: greeter
topics:
- singups: (1)
typeUrl: (2) "com.user.foo.bar.greeter.SingupMessage"
target: (3) "com.user.foo.bar/SingupProcessor"
{code}
Defines a Kafka ingress that consumes <utf8 strings, bytes > from a singups
topic,
and produces an Routable Protobuf message with the following type and
properties:
{code}
message Routable {
Address target; (1)
Any payload;
}
{code}
Where:
(1) is Address(FunctionType(com.user.foo.bar, SingupProcessor), <a consumer
record's key>)
(2) the Any's typeUrl would be com.user.foo.bar.greeter.SingupMessage and the
value bytes
would come directly from the consumer record value bytes
This would require an additional AutoRoutable router,
that basically forwards the payload to the target address.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
