[jira] [Commented] (FLINK-17928) Incorrect state size reported when using unaligned checkpoints

["Roman Khachatryan (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-17928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17116021#comment-17116021 ] 

Roman Khachatryan commented on FLINK-17928:
-------------------------------------------

The numbers were obtained using:

 
{code:java}
transferredBytes() {
 rest_url="http://`hostname`:20888/proxy/`get_aid`"
 job_id=`curl $rest_url/jobs/ | jq -r '.jobs[0].id'`
 vertex_ids=( `curl $rest_url/jobs/$job_id/ | jq -r '.vertices[].id'` )
 numBytesOut=0
 >&2 echo "job_id $job_id; vertex_ids: ${vertex_ids[@]}"
 for vertex in "${vertex_ids[@]}";
 do
 stats=`curl $rest_url/jobs/$job_id/vertices/$vertex/subtasks/metrics\?get=numBytesOut\&agg=sum`
 >&2 echo "vertex $vertex $stats"
 numBytesOut=$(($numBytesOut + `echo $stats | jq -s '[add | .[].sum] | add // 0' `))
 done
 echo "$numBytesOut" | numfmt --to=iec-i --suffix=B --padding=7
} 
{code}
 

> Incorrect state size reported when using unaligned checkpoints 
> ---------------------------------------------------------------
>
>                 Key: FLINK-17928
>                 URL: https://issues.apache.org/jira/browse/FLINK-17928
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / Checkpointing
>    Affects Versions: 1.11.0
>            Reporter: Piotr Nowojski
>            Priority: Critical
>             Fix For: 1.11.0
>
>
> Even when checkpoints on HDFS are between 100-300MBs, the reported state size is in orders of magnitude larger with values like:
> {noformat}
> 1GiB  1.5TiB  2.0TiB  2.1TiB  2.1TiB
> 148GiB  148GiB  148GiB  148GiB  148GiB  148GiB
> {noformat}
> it's probably because we have multiple {{Collection<InputChannelStateHandle>}}, and each of the individual handle returns the same value from {{AbstractChannelStateHandle#getStateSize}} - the full size of the spilled data, ignoring that only small portion of those data belong to a single input channel/result subpartition. In other words {{
> org.apache.flink.runtime.state.AbstractChannelStateHandle#getStateSize}} should be taking the offsets into account and return only the size of the data that belong exclusively to this handle.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)