[
https://issues.apache.org/jira/browse/FLINK-13721?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kerem Ulutaş updated FLINK-13721:
---------------------------------
Description:
Hi everyone,
Sorry if I'm doing anything wrong, this is my first issue in Apache Jira.
I have a use case which requires 2 data streams to be cross joined. To be
exact, one stream is location updates from clients and the other stream is
event data with location information. I'm trying to get events that happen
within a certain radius of client location(s).
Since the streams can not be related to each other by using a common key for
partitioning, I have to broadcast client updates to all tasks and evaluate the
radius check for each event.
The requirement here is, if we don't receive any location updates from a client
for a certain amount of time, we should consider the client is "gone" (similar
to the rationale stated in FLINK-3089 description)
I have attached the sample application classes I used to debug BroadcastState
and StateTTL together.
The output (see flink_broadcast_state_ttl_debug.log) shows that client "c0" got
its first event at 17:08:07.67 (expected to be evicted sometime after
17:08:37.xxx) but doesn't get evicted.
For the operator part (which is the result of
BroadcastConnectedStream<IntHolder, StringHolder>.process) - since context in
onTimer method doesn't let user to change contents of the broadcast state, only
way to deal with stale client data is as follows:
* In processElement method, calculate result for client data which is newer
than the ttl
* In processBroadcastElement method, remove client data older than a certain
amount of time from the broadcast state
If broadcast side of the connected streams doesn't get data for longer than the
desired time-to-live amount of time, BroadcastState will hold stale data and
processElement method would have to filter those client data each time. This is
the method I am using in production right now.
I am not aware if BroadcastState already supports StateTTL and I'm using it
incorrectly. I am also not aware of any decision or limitation that makes it
not possible to implement StateTTL for BroadcastState, I will be pleased if
someone explains if there are any.
Thanks and regards.
was:
Hi everyone,
Sorry if I'm doing anything wrong, this is my first issue in Apache Jira.
I have a use case which requires 2 data streams to be cross joined. To be
exact, one stream is location updates from clients and the other stream is
event data with location information. I'm trying to get events that happen
within a certain radius of client location(s).
Since the streams can not be related to each other by using a common key for
partitioning, I have to broadcast client updates to all tasks and evaluate the
radius check for each event.
The requirement here is, if we don't receive any location updates from a client
for a certain amount of time, we should consider the client is "gone" (similar
to the rationale stated in FLINK-3089 description)
I have attached the sample application classes I used to debug BroadcastState
and StateTTL together.
The output (see flink_broadcast_state_ttl_debug.log) shows that client "c0" got
its first event at 17:08:07.67 (expected to be evicted sometime after
17:08:37.xxx) but doesn't get evicted.
For the operator part (which is the result of
`BroadcastConnectedStream<IntHolder, StringHolder>.process`) - since context in
onTimer method doesn't let user to change contents of the broadcast state, only
way to deal with stale client data is as follows:
* In processElement method, calculate result for client data which is newer
than the ttl
* In processBroadcastElement method, remove client data older than a certain
amount of time from the broadcast state
If broadcast side of the connected streams doesn't get data for longer than the
desired time-to-live amount of time, BroadcastState will hold stale data and
processElement method would have to filter those client data each time. This is
the method I am using in production right now.
I am not aware if BroadcastState already supports StateTTL and I'm using it
incorrectly. I am also not aware of any decision or limitation that makes it
not possible to implement StateTTL for BroadcastState, I will be pleased if
someone explains if there are any.
Thanks and regards.
> BroadcastState should support StateTTL
> --------------------------------------
>
> Key: FLINK-13721
> URL: https://issues.apache.org/jira/browse/FLINK-13721
> Project: Flink
> Issue Type: Improvement
> Components: API / DataStream, Runtime / Queryable State
> Affects Versions: 1.8.1
> Environment: MacOS 10.14.6 running IntelliJ Idea Ultimate 2019.2,
> Flink version 1.8.1
> Reporter: Kerem Ulutaş
> Priority: Major
> Labels: feature, improvement, newbie
> Attachments: DebugBroadcastStateTTL.java, IntHolder.java,
> StringHolder.java, flink_broadcast_state_ttl_debug.log
>
>
> Hi everyone,
> Sorry if I'm doing anything wrong, this is my first issue in Apache Jira.
> I have a use case which requires 2 data streams to be cross joined. To be
> exact, one stream is location updates from clients and the other stream is
> event data with location information. I'm trying to get events that happen
> within a certain radius of client location(s).
> Since the streams can not be related to each other by using a common key for
> partitioning, I have to broadcast client updates to all tasks and evaluate
> the radius check for each event.
> The requirement here is, if we don't receive any location updates from a
> client for a certain amount of time, we should consider the client is "gone"
> (similar to the rationale stated in FLINK-3089 description)
> I have attached the sample application classes I used to debug BroadcastState
> and StateTTL together.
> The output (see flink_broadcast_state_ttl_debug.log) shows that client "c0"
> got its first event at 17:08:07.67 (expected to be evicted sometime after
> 17:08:37.xxx) but doesn't get evicted.
> For the operator part (which is the result of
> BroadcastConnectedStream<IntHolder, StringHolder>.process) - since context in
> onTimer method doesn't let user to change contents of the broadcast state,
> only way to deal with stale client data is as follows:
> * In processElement method, calculate result for client data which is newer
> than the ttl
> * In processBroadcastElement method, remove client data older than a certain
> amount of time from the broadcast state
> If broadcast side of the connected streams doesn't get data for longer than
> the desired time-to-live amount of time, BroadcastState will hold stale data
> and processElement method would have to filter those client data each time.
> This is the method I am using in production right now.
> I am not aware if BroadcastState already supports StateTTL and I'm using it
> incorrectly. I am also not aware of any decision or limitation that makes it
> not possible to implement StateTTL for BroadcastState, I will be pleased if
> someone explains if there are any.
> Thanks and regards.
--
This message was sent by Atlassian JIRA
(v7.6.14#76016)
