[
https://issues.apache.org/jira/browse/FLINK-26408?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hengyu Dai updated FLINK-26408:
-------------------------------
Description:
RetractableTopNFunction will throw a RuntimeException when
# the sorted Map {color:#0747a6}ValueState<SortedMap<RowData, Long>>
treeMap{color} is not empty.
# and the sorted Map doesn't contains current sort key.
Now we have Flink SQL job:
{code:java}
// table_a(a_key, a_time, a_jk), table_b(b_key, b_time, b_jk)
select
a_key,a_time,a_jk,b_key,b_time,b_jk
from
(
select
a_key,a_time,a_jk,b_key,b_time,b_jk,
row_number() over(partition by a_key order by a_time desc) as rn
from
(
select a_key, a_time, a_jk
from (
select * , row_number() over(partition by a_key order by a_time
desc) as rn
from table_a
) tmp1
where rn = 1
) t1
left join
(
select b_key, b_time, b_jk
from (
select * , row_number() over(partition by b_key order by b_time
desc) as rn
from table_b
) tmp2
where rn = 1
) t2
on t1.a_jk = t2.b_jk
) t3
where rn = 1{code}
the JobGraph is like:
{{Source table_a —> Rank_a}}
{{—> Join —>
Final Rank }}
{{Source table_b —> Rank_b}}
Suppose we hava following input:
#
||ts||SourceA
(a_key, a_time,a_jk)||SourceB
(b_key,b_time,b_jk)||RankA
(a_key, a_time,a_jk)||RankB
(b_key,b_time,b_jk)||Join
(a_key,b_key,a_time, a_jk)||Final Rank
(a_key,b_key,a_time)||
|t1| |+(b1,1,jk1)| |+(b1,1,jk1)| | |
|t2| |+(b2,2,jk2)| |+(b2,2,jk2)| | |
|t3|+(a1,3,jk1)| |+(a1,3,jk1)| |+(a1,b1,3,jk1)|+(a1,b1,3)|
|t4|+(a1,4,jk1)| |-(a1,3,jk1)
+(a1,4,jk1)| |-(a1,b1,3,jk1)
+(a1,b1,4,jk1)|-(a1,b1,3)
+(a1,b1,4)|
|t5|+(a1,5,jk2)| |-(a1,4,jk1)
+(a1,5,jk2)| |-(a1,b1,4,jk1)
+(a1,b2,5,jk2)|-(a1,b1,4)
+(a1,b2,5)|
| | | | | | | |
Assume:
# t4&t5 is almost at the same time, the Join Operator produce 4 message at
t4&t5, as the Hash Key changed(from jk1 to jk2), +(a1,b2,5,jk2) may runs on
different task from the other 3 messages, and it may arrrive earlier than them
# Due to network congestion or high machine load, etc. the messages produced
by t4&t5 take a while before they arrive Final Rank, when Final Rank received
them, the state is expired because of State TTL.
Now if +(a1,b2,5,jk2)arrives Final Rank first, the sortedMap of partition key
a1 will put a sort value 5. then when -(a1,b1,3,jk1)arrives Final Rank, it will
find that then sortedMap is not empty, and it doesn't contains sort key value
3. meet the conditions for that Runtime Exception.
we met this exception in our production environment, it's very serious because
when is happens, the job can not recover automatically as the state is polluted.
was:
RetractableTopNFunction will throw a RuntimeException when
# the sorted Map {{ValueState<SortedMap<RowData, Long>> treeMap }}is not empty.
# and the sorted Map doesn't contains current sort key.
Now we have Flink SQL job:
{code:java}
// table_a(a_key, a_time, a_jk), table_b(b_key, b_time, b_jk)
select
a_key,a_time,a_jk,b_key,b_time,b_jk
from
(
select
a_key,a_time,a_jk,b_key,b_time,b_jk,
row_number() over(partition by a_key order by a_time desc) as rn
from
(
select a_key, a_time, a_jk
from (
select * , row_number() over(partition by a_key order by a_time
desc) as rn
from table_a
) tmp1
where rn = 1
) t1
left join
(
select b_key, b_time, b_jk
from (
select * , row_number() over(partition by b_key order by b_time
desc) as rn
from table_b
) tmp2
where rn = 1
) t2
on t1.a_jk = t2.b_jk
) t3
where rn = 1{code}
the JobGraph is like:
{{Source table_a —> Rank_a}}
{{—> Join —>
Final Rank }}
{{Source table_b —> Rank_b}}
Suppose we hava following input:
#
||ts||SourceA
(a_key, a_time,a_jk)||SourceB
(b_key,b_time,b_jk)||RankA
(a_key, a_time,a_jk)||RankB
(b_key,b_time,b_jk)||Join
(a_key,b_key,a_time, a_jk)||Final Rank
(a_key,b_key,a_time)||
|t1| |+(b1,1,jk1)| |+(b1,1,jk1)| | |
|t2| |+(b2,2,jk2)| |+(b2,2,jk2)| | |
|t3|+(a1,3,jk1)| |+(a1,3,jk1)| |+(a1,b1,3,jk1)|+(a1,b1,3)|
|t4|+(a1,4,jk1)| |-(a1,3,jk1)
+(a1,4,jk1)| |-(a1,b1,3,jk1)
+(a1,b1,4,jk1)|-(a1,b1,3)
+(a1,b1,4)|
|t5|+(a1,5,jk2)| |-(a1,4,jk1)
+(a1,5,jk2)| |-(a1,b1,4,jk1)
+(a1,b2,5,jk2)|-(a1,b1,4)
+(a1,b2,5)|
| | | | | | | |
Assume:
# t4&t5 is almost at the same time, the Join Operator produce 4 message at
t4&t5, as the Hash Key changed(jk1 -> jk2), three of them {{-(a1,b1,3,jk1)
+(a1,b1,4,jk1) -(a1,b1,4,jk1) }}runs on task1, and the last one
{{+(a1,b2,5,jk2) }}maybe runs on another task2.
# Due to network congestion or high machine load, etc. the messages produced
by t4&t5 take a while before they arrive Final Rank, when Final Rank received
them, the state is expired because of State TTL.
Now if {{+(a1,b2,5,jk2) }}arrives Final Rank first, the sortedMap of partition
key a1 will put a sort value 5. then when {{-(a1,b1,3,jk1) }}arrives Final
Rank, it will find that then sortedMap is not empty, and it doesn't contains
sort key value 3. meet the conditions for that Runtime Exception.
we met this exception in our production environment, it's very serious because
when is happens, the job can not recover automatically as the state is polluted.
> retract a non-existent record in RetractableTopNFunction
> ---------------------------------------------------------
>
> Key: FLINK-26408
> URL: https://issues.apache.org/jira/browse/FLINK-26408
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner, Table SQL / Runtime
> Reporter: Hengyu Dai
> Priority: Major
>
> RetractableTopNFunction will throw a RuntimeException when
> # the sorted Map {color:#0747a6}ValueState<SortedMap<RowData, Long>>
> treeMap{color} is not empty.
> # and the sorted Map doesn't contains current sort key.
>
> Now we have Flink SQL job:
>
> {code:java}
> // table_a(a_key, a_time, a_jk), table_b(b_key, b_time, b_jk)
> select
> a_key,a_time,a_jk,b_key,b_time,b_jk
> from
> (
> select
> a_key,a_time,a_jk,b_key,b_time,b_jk,
> row_number() over(partition by a_key order by a_time desc) as rn
> from
> (
> select a_key, a_time, a_jk
> from (
> select * , row_number() over(partition by a_key order by a_time
> desc) as rn
> from table_a
> ) tmp1
> where rn = 1
> ) t1
> left join
> (
> select b_key, b_time, b_jk
> from (
> select * , row_number() over(partition by b_key order by b_time
> desc) as rn
> from table_b
> ) tmp2
> where rn = 1
> ) t2
> on t1.a_jk = t2.b_jk
> ) t3
> where rn = 1{code}
> the JobGraph is like:
> {{Source table_a —> Rank_a}}
> {{—> Join —>
> Final Rank }}
> {{Source table_b —> Rank_b}}
>
>
>
> Suppose we hava following input:
>
> #
> ||ts||SourceA
> (a_key, a_time,a_jk)||SourceB
> (b_key,b_time,b_jk)||RankA
> (a_key, a_time,a_jk)||RankB
> (b_key,b_time,b_jk)||Join
> (a_key,b_key,a_time, a_jk)||Final Rank
> (a_key,b_key,a_time)||
> |t1| |+(b1,1,jk1)| |+(b1,1,jk1)| | |
> |t2| |+(b2,2,jk2)| |+(b2,2,jk2)| | |
> |t3|+(a1,3,jk1)| |+(a1,3,jk1)| |+(a1,b1,3,jk1)|+(a1,b1,3)|
> |t4|+(a1,4,jk1)| |-(a1,3,jk1)
> +(a1,4,jk1)| |-(a1,b1,3,jk1)
> +(a1,b1,4,jk1)|-(a1,b1,3)
> +(a1,b1,4)|
> |t5|+(a1,5,jk2)| |-(a1,4,jk1)
> +(a1,5,jk2)| |-(a1,b1,4,jk1)
> +(a1,b2,5,jk2)|-(a1,b1,4)
> +(a1,b2,5)|
> | | | | | | | |
> Assume:
>
> # t4&t5 is almost at the same time, the Join Operator produce 4 message at
> t4&t5, as the Hash Key changed(from jk1 to jk2), +(a1,b2,5,jk2) may runs on
> different task from the other 3 messages, and it may arrrive earlier than them
> # Due to network congestion or high machine load, etc. the messages produced
> by t4&t5 take a while before they arrive Final Rank, when Final Rank received
> them, the state is expired because of State TTL.
> Now if +(a1,b2,5,jk2)arrives Final Rank first, the sortedMap of partition key
> a1 will put a sort value 5. then when -(a1,b1,3,jk1)arrives Final Rank, it
> will find that then sortedMap is not empty, and it doesn't contains sort key
> value 3. meet the conditions for that Runtime Exception.
> we met this exception in our production environment, it's very serious
> because when is happens, the job can not recover automatically as the state
> is polluted.
>
>
>
>
>
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
