You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/03/22 08:52:49 UTC

[GitHub] [flink] Thesharing opened a new pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Thesharing opened a new pull request #15314:
URL: https://github.com/apache/flink/pull/15314


   ## What is the purpose of the change
   
   *This pull request introduce the optimization of releasing result partitions in RegionPartitionReleaseStrategy.*
   *RegionPartitionReleaseStrategy is responsible for releasing result partitions when all the downstream tasks finish.*
   
   *The current implementation is:*
   ```
   for each consumed SchedulingResultPartition of current finished SchedulingPipelinedRegion:
     for each consumer SchedulingPipelinedRegion of the SchedulingResultPartition:
       if all the regions are finished:
         release the partitions
   ```
   
   *The time complexity of releasing a result partition is O(N^2). However, considering that during the entire stage, all the result partitions need to be released, the time complexity is actually O(N^3).*
   
   *Based on FLINK-21228 and FLINK-21330, the consumed result partitions of a pipelined region are grouped. Since the result partitions in one group are isomorphic, we can just cache the finished status of the pipeline regions and the fully consumed status of result partition groups.*
   
   *The optimized implementation is:*
   ```
   for each ConsumedPartitionGroup of current finished SchedulingPipelinedRegion:
     if all consumer SchedulingPipelinedRegion of the ConsumedPartitionGroup are finished:
       set the ConsumePartitionGroup to be fully consumed
       for result partition in the ConsumePartitionGroup:
         if all the ConsumePartitionGroups it belongs to are fully consumed:
           release the result partition
   ```
   
   *After the optimization, the complexity decreases from O(N^3) to O(N).*
   
   *For more details, please check FLINK-21332.*
   
   ## Brief change log
   
     - *Optimize RegionPartitionReleaseStrategy#filterReleasablePartitions*
   
   
   ## Verifying this change
   
   *Since this optimization does not change the original logic of eleasing result partitions in RegionPartitionReleaseStrategy, we believe that this change is already covered by RegionPartitionReleaseStrategyTest.*
   
   ## Does this pull request potentially affect one of the following parts:
   
     - Dependencies (does it add or upgrade a dependency): (yes / **no**)
     - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
     - The serializers: (yes / **no** / don't know)
     - The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
     - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (**yes** / no / don't know)
     - The S3 file system connector: (yes / **no** / don't know)
   
   ## Documentation
   
     - Does this pull request introduce a new feature? (yes / **no**)
     - If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2dc45dbffe9740c0af706ff2c0ce6d241620be9e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422) 
   * d69e745e1e5a11209d8eacc00281df878915e0b9 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c101e777cbdd1a2ec055ec62901be25d24da2f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "CANCELED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * d69e745e1e5a11209d8eacc00281df878915e0b9 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452) 
   * 2ec0121bc9dd5235afb2b7de48ffac3df012ff48 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     }, {
       "hash" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713",
       "triggerID" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "triggerType" : "PUSH"
     }, {
       "hash" : "f6a937c54000be142978fc15aab23bb8fb33f701",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15750",
       "triggerID" : "f6a937c54000be142978fc15aab23bb8fb33f701",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   * f6a937c54000be142978fc15aab23bb8fb33f701 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15750) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 7512eec2a3d84fb8211cf9b1161a25b7e8b8800e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380) 
   * 2dc45dbffe9740c0af706ff2c0ce6d241620be9e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803885660


   Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
   to review your pull request. We will use this comment to track the progress of the review.
   
   
   ## Automated Checks
   Last check on commit f6a937c54000be142978fc15aab23bb8fb33f701 (Sat Aug 28 11:09:01 UTC 2021)
   
   **Warnings:**
    * No documentation files were touched! Remember to keep the Flink docs up to date!
   
   
   <sub>Mention the bot in a comment to re-run the automated checks.</sub>
   ## Review Progress
   
   * ❓ 1. The [description] looks good.
   * ❓ 2. There is [consensus] that the contribution should go into to Flink.
   * ❓ 3. Needs [attention] from.
   * ❓ 4. The change fits into the overall [architecture].
   * ❓ 5. Overall code [quality] is good.
   
   Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
    The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
    - `@flinkbot approve all` to approve all aspects
    - `@flinkbot approve-until architecture` to approve everything until `architecture`
    - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
    - `@flinkbot disapprove architecture` to remove an approval you gave earlier
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 888160af8cc418e0e4efb2d07de906d55345d6cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     }, {
       "hash" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713",
       "triggerID" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c101e777cbdd1a2ec055ec62901be25d24da2f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691) 
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   * cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 888160af8cc418e0e4efb2d07de906d55345d6cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172) 
   * 7512eec2a3d84fb8211cf9b1161a25b7e8b8800e UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c101e777cbdd1a2ec055ec62901be25d24da2f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691) 
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 7512eec2a3d84fb8211cf9b1161a25b7e8b8800e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380) 
   * 2dc45dbffe9740c0af706ff2c0ce6d241620be9e UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * d69e745e1e5a11209d8eacc00281df878915e0b9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452) 
   * 2ec0121bc9dd5235afb2b7de48ffac3df012ff48 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 7512eec2a3d84fb8211cf9b1161a25b7e8b8800e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-804025086


   It's worth noting that although after the optimization the complexity is O(N^2), the time cost on RegionPartitionReleaseStrategy#filterReleasablePartitions is still a bit long. As illustrated in the figure below, we can see that most of time is spent on `HashMap.get`.
   
   ![Illustration](https://user-images.githubusercontent.com/6576831/111989779-2e45c880-8b4d-11eb-93e8-ed2ecb31729a.png)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhuzhurk commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
zhuzhurk commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r602050713



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>
+            executionViewByRegion = new HashMap<>();
+
+    void init(Iterable<ConsumerRegionGroupExecutionView> executionViews) {

Review comment:
       let do it directly in the constructor

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
##########
@@ -47,10 +49,26 @@
     private final Map<ExecutionVertexID, PipelinedRegionExecutionView> regionExecutionViewByVertex =
             new HashMap<>();
 
+    private final Map<ConsumedPartitionGroup, ConsumerRegionGroupExecutionView>
+            partitionGroupConsumerRegions = new HashMap<>();
+
+    private final Map<SchedulingPipelinedRegion, Set<ConsumedPartitionGroup>>
+            regionConsumedPartitionGroups = new HashMap<>();
+
+    private final ConsumerRegionGroupExecutionViewTracker consumerRegionGroupExecutionViewTracker =
+            new ConsumerRegionGroupExecutionViewTracker();
+
     public RegionPartitionReleaseStrategy(final SchedulingTopology schedulingTopology) {
         this.schedulingTopology = checkNotNull(schedulingTopology);
+        init();
+    }
 
+    private void init() {

Review comment:
       I feel that there is no need to separate this `init()` method.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>
+            executionViewByRegion = new HashMap<>();
+
+    void init(Iterable<ConsumerRegionGroupExecutionView> executionViews) {
+        for (ConsumerRegionGroupExecutionView executionView : executionViews) {
+            for (SchedulingPipelinedRegion region : executionView) {
+                executionViewByRegion
+                        .computeIfAbsent(region, r -> new ArrayList<>())
+                        .add(executionView);
+            }
+        }
+    }
+
+    void regionFinished(SchedulingPipelinedRegion region) {
+        for (ConsumerRegionGroupExecutionView executionView :
+                executionViewByRegion.getOrDefault(region, Collections.emptyList())) {
+            executionView.regionFinished(region);
+        }
+    }
+
+    void regionUnfinished(SchedulingPipelinedRegion region) {
+        for (ConsumerRegionGroupExecutionView executionView :
+                executionViewByRegion.getOrDefault(region, Collections.emptyList())) {

Review comment:
       I would check to ensure that `executionViewByRegion.get(region)` is not null or empty 

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionView.java
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * This view maintains the finished progress of consumer {@link SchedulingPipelinedRegion}s for each
+ * {@link ConsumedPartitionGroup}.
+ */
+public class ConsumerRegionGroupExecutionView implements Iterable<SchedulingPipelinedRegion> {

Review comment:
       I hope we can also have a test for this class.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTrackerTest.java
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.adapter.DefaultSchedulingPipelinedRegion;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingTopology;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link ConsumerRegionGroupExecutionView} and {@link
+ * ConsumerRegionGroupExecutionViewTracker}.
+ */
+public class ConsumerRegionGroupExecutionViewTrackerTest extends TestLogger {
+
+    List<TestingSchedulingExecutionVertex> producers;
+    List<TestingSchedulingExecutionVertex> consumers;
+
+    TestingSchedulingTopology schedulingTopology;

Review comment:
       This topology is not needed.
   One can directly create some individual `TestingSchedulingPipelinedRegion`s and use them to build some `ConsumerRegionGroupExecutionView`s and then create the tracker. This further allows to simplifies these tests.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>
+            executionViewByRegion = new HashMap<>();
+
+    void init(Iterable<ConsumerRegionGroupExecutionView> executionViews) {
+        for (ConsumerRegionGroupExecutionView executionView : executionViews) {
+            for (SchedulingPipelinedRegion region : executionView) {
+                executionViewByRegion
+                        .computeIfAbsent(region, r -> new ArrayList<>())
+                        .add(executionView);
+            }
+        }
+    }
+
+    void regionFinished(SchedulingPipelinedRegion region) {
+        for (ConsumerRegionGroupExecutionView executionView :

Review comment:
       I would check to ensure that `executionViewByRegion.get(region)` is not null or empty instead of creating an empty list

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
##########
@@ -98,26 +158,24 @@ private PipelinedRegionExecutionView getPipelinedRegionExecutionViewForVertex(
     }
 
     private List<IntermediateResultPartitionID> filterReleasablePartitions(
-            final Iterable<? extends SchedulingResultPartition> schedulingResultPartitions) {
-        return IterableUtils.toStream(schedulingResultPartitions)
-                .map(SchedulingResultPartition::getId)
-                .filter(this::areConsumerRegionsFinished)
-                .collect(Collectors.toList());
-    }
-
-    private boolean areConsumerRegionsFinished(
-            final IntermediateResultPartitionID resultPartitionId) {
-        final SchedulingResultPartition resultPartition =
-                schedulingTopology.getResultPartition(resultPartitionId);
-        return IterableUtils.toStream(resultPartition.getConsumers())
-                .map(SchedulingExecutionVertex::getId)
-                .allMatch(this::isRegionOfVertexFinished);
-    }
+            final Iterable<ConsumedPartitionGroup> consumedPartitionGroups) {
+
+        final List<IntermediateResultPartitionID> filteredPartitions = new ArrayList<>();
+
+        for (ConsumedPartitionGroup consumedPartitionGroup : consumedPartitionGroups) {
+            final ConsumerRegionGroupExecutionView consumerRegionSet =

Review comment:
       nit: consumerRegionSet -> consumerRegionGroup

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {

Review comment:
       Current the role of this class is to maintain the status of `ConsumerRegionGroupExecutionView` instead of tracking them, so I would rename it to `ConsumerRegionGroupExecutionViewMaintainer`




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r603259955



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionView.java
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * This view maintains the finished progress of consumer {@link SchedulingPipelinedRegion}s for each
+ * {@link ConsumedPartitionGroup}.
+ */
+public class ConsumerRegionGroupExecutionView implements Iterable<SchedulingPipelinedRegion> {

Review comment:
       I believe that `ConsumerRegionExecutionView` is covered by `ConsumerRegionExecutionViewMaintainerTest`.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
##########
@@ -47,10 +49,26 @@
     private final Map<ExecutionVertexID, PipelinedRegionExecutionView> regionExecutionViewByVertex =
             new HashMap<>();
 
+    private final Map<ConsumedPartitionGroup, ConsumerRegionGroupExecutionView>
+            partitionGroupConsumerRegions = new HashMap<>();
+
+    private final Map<SchedulingPipelinedRegion, Set<ConsumedPartitionGroup>>
+            regionConsumedPartitionGroups = new HashMap<>();
+
+    private final ConsumerRegionGroupExecutionViewTracker consumerRegionGroupExecutionViewTracker =
+            new ConsumerRegionGroupExecutionViewTracker();
+
     public RegionPartitionReleaseStrategy(final SchedulingTopology schedulingTopology) {
         this.schedulingTopology = checkNotNull(schedulingTopology);
+        init();
+    }
 
+    private void init() {

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>
+            executionViewByRegion = new HashMap<>();
+
+    void init(Iterable<ConsumerRegionGroupExecutionView> executionViews) {

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>
+            executionViewByRegion = new HashMap<>();
+
+    void init(Iterable<ConsumerRegionGroupExecutionView> executionViews) {
+        for (ConsumerRegionGroupExecutionView executionView : executionViews) {
+            for (SchedulingPipelinedRegion region : executionView) {
+                executionViewByRegion
+                        .computeIfAbsent(region, r -> new ArrayList<>())
+                        .add(executionView);
+            }
+        }
+    }
+
+    void regionFinished(SchedulingPipelinedRegion region) {
+        for (ConsumerRegionGroupExecutionView executionView :

Review comment:
       Since ConsumerRegionGroupExecutionViewMaintainer only maintains the status of regions with ConsumedPartitionGroup, we need to use `getOrDefault` to avoid Exceptions for source regions.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTrackerTest.java
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.adapter.DefaultSchedulingPipelinedRegion;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingTopology;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link ConsumerRegionGroupExecutionView} and {@link
+ * ConsumerRegionGroupExecutionViewTracker}.
+ */
+public class ConsumerRegionGroupExecutionViewTrackerTest extends TestLogger {
+
+    List<TestingSchedulingExecutionVertex> producers;
+    List<TestingSchedulingExecutionVertex> consumers;
+
+    TestingSchedulingTopology schedulingTopology;

Review comment:
       Agreed. I've simplified this test. Please re-review it.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
##########
@@ -98,26 +158,24 @@ private PipelinedRegionExecutionView getPipelinedRegionExecutionViewForVertex(
     }
 
     private List<IntermediateResultPartitionID> filterReleasablePartitions(
-            final Iterable<? extends SchedulingResultPartition> schedulingResultPartitions) {
-        return IterableUtils.toStream(schedulingResultPartitions)
-                .map(SchedulingResultPartition::getId)
-                .filter(this::areConsumerRegionsFinished)
-                .collect(Collectors.toList());
-    }
-
-    private boolean areConsumerRegionsFinished(
-            final IntermediateResultPartitionID resultPartitionId) {
-        final SchedulingResultPartition resultPartition =
-                schedulingTopology.getResultPartition(resultPartitionId);
-        return IterableUtils.toStream(resultPartition.getConsumers())
-                .map(SchedulingExecutionVertex::getId)
-                .allMatch(this::isRegionOfVertexFinished);
-    }
+            final Iterable<ConsumedPartitionGroup> consumedPartitionGroups) {
+
+        final List<IntermediateResultPartitionID> filteredPartitions = new ArrayList<>();
+
+        for (ConsumedPartitionGroup consumedPartitionGroup : consumedPartitionGroups) {
+            final ConsumerRegionGroupExecutionView consumerRegionSet =

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewTracker.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewTracker {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>
+            executionViewByRegion = new HashMap<>();
+
+    void init(Iterable<ConsumerRegionGroupExecutionView> executionViews) {
+        for (ConsumerRegionGroupExecutionView executionView : executionViews) {
+            for (SchedulingPipelinedRegion region : executionView) {
+                executionViewByRegion
+                        .computeIfAbsent(region, r -> new ArrayList<>())
+                        .add(executionView);
+            }
+        }
+    }
+
+    void regionFinished(SchedulingPipelinedRegion region) {
+        for (ConsumerRegionGroupExecutionView executionView :
+                executionViewByRegion.getOrDefault(region, Collections.emptyList())) {
+            executionView.regionFinished(region);
+        }
+    }
+
+    void regionUnfinished(SchedulingPipelinedRegion region) {
+        for (ConsumerRegionGroupExecutionView executionView :
+                executionViewByRegion.getOrDefault(region, Collections.emptyList())) {

Review comment:
       The same above.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhuzhurk commented on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
zhuzhurk commented on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-807910795


   > It's worth noting that although after the optimization the complexity is O(N), the time cost on RegionPartitionReleaseStrategy#filterReleasablePartitions is still a bit long. As illustrated in the figure below, we can see that most of time is spent on `HashMap.get`.
   > 
   > ![Illustration](https://user-images.githubusercontent.com/6576831/111989779-2e45c880-8b4d-11eb-93e8-ed2ecb31729a.png)
   
   I would suggest to introduce the assumption that one `IntermediateResultPartition` can have one only `ConsumerVertexGroup`(indicating that one `IntermediateDataSet` can have one only consumer `JobEdge` which is already a widely assumption in flink-runtime at the moment). This can help to reduce the complexity for each `vertexFinished()` invocation to O(1).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 888160af8cc418e0e4efb2d07de906d55345d6cb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172) 
   * 7512eec2a3d84fb8211cf9b1161a25b7e8b8800e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2ec0121bc9dd5235afb2b7de48ffac3df012ff48 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508) 
   * 2c101e777cbdd1a2ec055ec62901be25d24da2f9 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     }, {
       "hash" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713",
       "triggerID" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "triggerType" : "PUSH"
     }, {
       "hash" : "f6a937c54000be142978fc15aab23bb8fb33f701",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "f6a937c54000be142978fc15aab23bb8fb33f701",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   * cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713) 
   * f6a937c54000be142978fc15aab23bb8fb33f701 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhuzhurk commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
zhuzhurk commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r603273083



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
##########
@@ -473,7 +473,7 @@ public void setInitialState(@Nullable JobManagerTaskRestore taskRestore) {
     private static int getPartitionMaxParallelism(
             IntermediateResultPartition partition,
             Function<ExecutionVertexID, ExecutionVertex> getVertexById) {
-        final List<ConsumerVertexGroup> consumers = partition.getConsumers();
+        final List<ConsumerVertexGroup> consumers = partition.getConsumerVertexGroups();

Review comment:
       consumers -> consumerVertexGroups

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/PartitionDescriptor.java
##########
@@ -113,7 +113,7 @@ public static PartitionDescriptor from(IntermediateResultPartition partition) {
         // If no consumers are known at this point, we use a single subpartition, otherwise we have
         // one for each consuming sub task.
         int numberOfSubpartitions = 1;
-        List<ConsumerVertexGroup> consumers = partition.getConsumers();
+        List<ConsumerVertexGroup> consumers = partition.getConsumerVertexGroups();

Review comment:
       `consumers` -> `consumerVertexGroups`

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
##########
@@ -696,7 +696,7 @@ else if (current == RUNNING || current == DEPLOYING) {
 
     private void updatePartitionConsumers(final IntermediateResultPartition partition) {
 
-        final List<ConsumerVertexGroup> allConsumers = partition.getConsumers();
+        final List<ConsumerVertexGroup> allConsumers = partition.getConsumerVertexGroups();

Review comment:
       allConsumers -> consumerVertexGroups

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
##########
@@ -501,7 +501,7 @@ public static void verifyGeneratedExecutionJobVertex(
                 assertEquals(inputJobVertices.size(), ev.getNumberOfInputs());
 
                 for (int i = 0; i < inputJobVertices.size(); i++) {
-                    ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(i);
+                    ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(i);

Review comment:
       `consumedPartitions` -> `consumedPartitionGroup`

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -276,7 +276,7 @@ private void testConnections(
         for (int vertexIndex = 0; vertexIndex < target.getTaskVertices().length; vertexIndex++) {
 
             ExecutionVertex ev = target.getTaskVertices()[vertexIndex];
-            ConsumedPartitionGroup partitionIds = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup partitionIds = ev.getConsumedPartitionGroup(0);

Review comment:
       consumedPartitions  -> consumedPartitionGroup

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -192,7 +192,7 @@ private void testLowToHigh(int lowDop, int highDop) throws Exception {
         for (ExecutionVertex ev : target.getTaskVertices()) {
             assertEquals(1, ev.getNumberOfInputs());
 
-            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(0);

Review comment:
       consumedPartitions  -> consumedPartitionGroup

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -129,7 +129,7 @@ public void testNTo7N() throws Exception {
         for (ExecutionVertex ev : target.getTaskVertices()) {
             assertEquals(1, ev.getNumberOfInputs());
 
-            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(0);

Review comment:
       consumedPartitions  -> consumedPartitionGroup

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -111,7 +111,7 @@ public void testNTo2N() throws Exception {
         for (ExecutionVertex ev : target.getTaskVertices()) {
             assertEquals(1, ev.getNumberOfInputs());
 
-            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(0);

Review comment:
       consumedPartitions  -> consumedPartitionGroup




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 3764ec2edda2138751e0a3f61ad5e4077894d8f8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2ec0121bc9dd5235afb2b7de48ffac3df012ff48 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508) 
   * 2c101e777cbdd1a2ec055ec62901be25d24da2f9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-807912254


   > I would suggest to introduce the assumption that one `IntermediateResultPartition` can have one only `ConsumerVertexGroup`(indicating that one `IntermediateDataSet` can have one only consumer `JobEdge` which is already a widely assumption in flink-runtime at the moment). This can help to reduce the complexity for each `vertexFinished()` invocation to O(1).
   
   Thanks for proposing this solution 👍  I've already add the check and comments according to this assumption. Would you mind re-reviewing it once you got any free time? 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2dc45dbffe9740c0af706ff2c0ce6d241620be9e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r603763928



##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewMaintainerTest.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingPipelinedRegion;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link ConsumerRegionGroupExecutionView} and {@link
+ * ConsumerRegionGroupExecutionViewMaintainer}.
+ */
+public class ConsumerRegionGroupExecutionViewMaintainerTest extends TestLogger {
+
+    TestingSchedulingExecutionVertex producer;

Review comment:
       Thanks for reminding. Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewMaintainer.java
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewMaintainer {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>

Review comment:
       Resolved.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewMaintainerTest.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingPipelinedRegion;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link ConsumerRegionGroupExecutionView} and {@link
+ * ConsumerRegionGroupExecutionViewMaintainer}.
+ */
+public class ConsumerRegionGroupExecutionViewMaintainerTest extends TestLogger {
+
+    TestingSchedulingExecutionVertex producer;
+    TestingSchedulingExecutionVertex consumer;
+
+    TestingSchedulingPipelinedRegion producerRegion;
+    TestingSchedulingPipelinedRegion consumerRegion;
+
+    ConsumerRegionGroupExecutionView consumerRegionGroupExecutionView;
+    ConsumerRegionGroupExecutionViewMaintainer consumerRegionGroupExecutionViewMaintainer;
+
+    @Before
+    public void setup() {
+        createProducerAndConsumer();
+        createConsumerRegionGroupExecutionViewTracker();
+    }
+
+    @Test
+    public void testRegionFinished() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        assertTrue(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testRegionUnfinished() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(consumerRegion);
+
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testRegionFinishedMultipleTimes() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+
+        assertTrue(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testRegionUnfinishedMultipleTimes() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(consumerRegion);
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(consumerRegion);
+
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        assertTrue(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testFinishWrongRegion() {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(producerRegion);
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testUnfinishedWrongRegion() {
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(producerRegion);
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    private void createProducerAndConsumer() {
+        producer = TestingSchedulingExecutionVertex.newBuilder().build();
+        consumer = TestingSchedulingExecutionVertex.newBuilder().build();
+
+        producerRegion = new TestingSchedulingPipelinedRegion(Collections.singleton(producer));
+        consumerRegion = new TestingSchedulingPipelinedRegion(Collections.singleton(consumer));
+    }
+
+    private void createConsumerRegionGroupExecutionViewTracker() {

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
##########
@@ -101,27 +128,24 @@ private PipelinedRegionExecutionView getPipelinedRegionExecutionViewForVertex(
     }
 
     private List<IntermediateResultPartitionID> filterReleasablePartitions(
-            final Iterable<? extends SchedulingResultPartition> schedulingResultPartitions) {
-        return IterableUtils.toStream(schedulingResultPartitions)
-                .distinct()
-                .map(SchedulingResultPartition::getId)
-                .filter(this::areConsumerRegionsFinished)
-                .collect(Collectors.toList());
-    }
-
-    private boolean areConsumerRegionsFinished(
-            final IntermediateResultPartitionID resultPartitionId) {
-        final SchedulingResultPartition resultPartition =
-                schedulingTopology.getResultPartition(resultPartitionId);
-        return IterableUtils.toStream(resultPartition.getConsumers())
-                .map(SchedulingExecutionVertex::getId)
-                .allMatch(this::isRegionOfVertexFinished);
-    }
+            final Iterable<ConsumedPartitionGroup> consumedPartitionGroups) {
+
+        final List<IntermediateResultPartitionID> filteredPartitions = new ArrayList<>();

Review comment:
       Resolved.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "CANCELED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 3764ec2edda2138751e0a3f61ad5e4077894d8f8 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162) 
   * 888160af8cc418e0e4efb2d07de906d55345d6cb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * d69e745e1e5a11209d8eacc00281df878915e0b9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot commented on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803885660


   Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
   to review your pull request. We will use this comment to track the progress of the review.
   
   
   ## Automated Checks
   Last check on commit 3764ec2edda2138751e0a3f61ad5e4077894d8f8 (Mon Mar 22 08:56:08 UTC 2021)
   
   **Warnings:**
    * No documentation files were touched! Remember to keep the Flink docs up to date!
   
   
   <sub>Mention the bot in a comment to re-run the automated checks.</sub>
   ## Review Progress
   
   * ❓ 1. The [description] looks good.
   * ❓ 2. There is [consensus] that the contribution should go into to Flink.
   * ❓ 3. Needs [attention] from.
   * ❓ 4. The change fits into the overall [architecture].
   * ❓ 5. Overall code [quality] is good.
   
   Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
    The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
    - `@flinkbot approve all` to approve all aspects
    - `@flinkbot approve-until architecture` to approve everything until `architecture`
    - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
    - `@flinkbot disapprove architecture` to remove an approval you gave earlier
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2dc45dbffe9740c0af706ff2c0ce6d241620be9e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422) 
   * d69e745e1e5a11209d8eacc00281df878915e0b9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot commented on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 3764ec2edda2138751e0a3f61ad5e4077894d8f8 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhuzhurk closed pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
zhuzhurk closed pull request #15314:
URL: https://github.com/apache/flink/pull/15314


   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * d69e745e1e5a11209d8eacc00281df878915e0b9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452) 
   * 2ec0121bc9dd5235afb2b7de48ffac3df012ff48 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 3764ec2edda2138751e0a3f61ad5e4077894d8f8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162) 
   * 888160af8cc418e0e4efb2d07de906d55345d6cb UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2ec0121bc9dd5235afb2b7de48ffac3df012ff48 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r603498344



##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
##########
@@ -501,7 +501,7 @@ public static void verifyGeneratedExecutionJobVertex(
                 assertEquals(inputJobVertices.size(), ev.getNumberOfInputs());
 
                 for (int i = 0; i < inputJobVertices.size(); i++) {
-                    ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(i);
+                    ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(i);

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
##########
@@ -473,7 +473,7 @@ public void setInitialState(@Nullable JobManagerTaskRestore taskRestore) {
     private static int getPartitionMaxParallelism(
             IntermediateResultPartition partition,
             Function<ExecutionVertexID, ExecutionVertex> getVertexById) {
-        final List<ConsumerVertexGroup> consumers = partition.getConsumers();
+        final List<ConsumerVertexGroup> consumers = partition.getConsumerVertexGroups();

Review comment:
       Resolved.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -129,7 +129,7 @@ public void testNTo7N() throws Exception {
         for (ExecutionVertex ev : target.getTaskVertices()) {
             assertEquals(1, ev.getNumberOfInputs());
 
-            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(0);

Review comment:
       Resolved.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -111,7 +111,7 @@ public void testNTo2N() throws Exception {
         for (ExecutionVertex ev : target.getTaskVertices()) {
             assertEquals(1, ev.getNumberOfInputs());
 
-            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(0);

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
##########
@@ -696,7 +696,7 @@ else if (current == RUNNING || current == DEPLOYING) {
 
     private void updatePartitionConsumers(final IntermediateResultPartition partition) {
 
-        final List<ConsumerVertexGroup> allConsumers = partition.getConsumers();
+        final List<ConsumerVertexGroup> allConsumers = partition.getConsumerVertexGroups();

Review comment:
       Resolved.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -276,7 +276,7 @@ private void testConnections(
         for (int vertexIndex = 0; vertexIndex < target.getTaskVertices().length; vertexIndex++) {
 
             ExecutionVertex ev = target.getTaskVertices()[vertexIndex];
-            ConsumedPartitionGroup partitionIds = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup partitionIds = ev.getConsumedPartitionGroup(0);

Review comment:
       Resolved.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/PartitionDescriptor.java
##########
@@ -113,7 +113,7 @@ public static PartitionDescriptor from(IntermediateResultPartition partition) {
         // If no consumers are known at this point, we use a single subpartition, otherwise we have
         // one for each consuming sub task.
         int numberOfSubpartitions = 1;
-        List<ConsumerVertexGroup> consumers = partition.getConsumers();
+        List<ConsumerVertexGroup> consumers = partition.getConsumerVertexGroups();

Review comment:
       Resolved.

##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
##########
@@ -192,7 +192,7 @@ private void testLowToHigh(int lowDop, int highDop) throws Exception {
         for (ExecutionVertex ev : target.getTaskVertices()) {
             assertEquals(1, ev.getNumberOfInputs());
 
-            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitions(0);
+            ConsumedPartitionGroup consumedPartitions = ev.getConsumedPartitionGroup(0);

Review comment:
       Resolved.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     }, {
       "hash" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c101e777cbdd1a2ec055ec62901be25d24da2f9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691) 
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   * cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-804025086


   It's worth noting that although after the optimization the complexity is O(N), the time cost on RegionPartitionReleaseStrategy#filterReleasablePartitions is still a bit long. As illustrated in the figure below, we can see that most of time is spent on `HashMap.get`.
   
   ![Illustration](https://user-images.githubusercontent.com/6576831/111989779-2e45c880-8b4d-11eb-93e8-ed2ecb31729a.png)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-807905349


   @flinkbot run azure


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Thesharing commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
Thesharing commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r603748621



##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewMaintainerTest.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingPipelinedRegion;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link ConsumerRegionGroupExecutionView} and {@link
+ * ConsumerRegionGroupExecutionViewMaintainer}.
+ */
+public class ConsumerRegionGroupExecutionViewMaintainerTest extends TestLogger {
+
+    TestingSchedulingExecutionVertex producer;
+    TestingSchedulingExecutionVertex consumer;
+
+    TestingSchedulingPipelinedRegion producerRegion;
+    TestingSchedulingPipelinedRegion consumerRegion;
+
+    ConsumerRegionGroupExecutionView consumerRegionGroupExecutionView;
+    ConsumerRegionGroupExecutionViewMaintainer consumerRegionGroupExecutionViewMaintainer;
+
+    @Before
+    public void setup() {
+        createProducerAndConsumer();
+        createConsumerRegionGroupExecutionViewTracker();
+    }
+
+    @Test
+    public void testRegionFinished() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        assertTrue(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testRegionUnfinished() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(consumerRegion);
+
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testRegionFinishedMultipleTimes() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+
+        assertTrue(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testRegionUnfinishedMultipleTimes() throws Exception {
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(consumerRegion);
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(consumerRegion);
+
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(consumerRegion);
+        assertTrue(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testFinishWrongRegion() {
+        consumerRegionGroupExecutionViewMaintainer.regionFinished(producerRegion);
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    @Test
+    public void testUnfinishedWrongRegion() {
+        consumerRegionGroupExecutionViewMaintainer.regionUnfinished(producerRegion);
+        assertFalse(consumerRegionGroupExecutionView.isFinished());
+    }
+
+    private void createProducerAndConsumer() {
+        producer = TestingSchedulingExecutionVertex.newBuilder().build();
+        consumer = TestingSchedulingExecutionVertex.newBuilder().build();
+
+        producerRegion = new TestingSchedulingPipelinedRegion(Collections.singleton(producer));
+        consumerRegion = new TestingSchedulingPipelinedRegion(Collections.singleton(consumer));
+    }
+
+    private void createConsumerRegionGroupExecutionViewTracker() {

Review comment:
       ```suggestion
       private void createConsumerRegionGroupExecutionViewMaintainer() {
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     } ]
   }-->
   ## CI report:
   
   * d69e745e1e5a11209d8eacc00281df878915e0b9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     }, {
       "hash" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713",
       "triggerID" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   * cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15314:
URL: https://github.com/apache/flink/pull/15314#issuecomment-803903444


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15162",
       "triggerID" : "3764ec2edda2138751e0a3f61ad5e4077894d8f8",
       "triggerType" : "PUSH"
     }, {
       "hash" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15172",
       "triggerID" : "888160af8cc418e0e4efb2d07de906d55345d6cb",
       "triggerType" : "PUSH"
     }, {
       "hash" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15380",
       "triggerID" : "7512eec2a3d84fb8211cf9b1161a25b7e8b8800e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15422",
       "triggerID" : "2dc45dbffe9740c0af706ff2c0ce6d241620be9e",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15503",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "d69e745e1e5a11209d8eacc00281df878915e0b9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15452",
       "triggerID" : "807905349",
       "triggerType" : "MANUAL"
     }, {
       "hash" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15508",
       "triggerID" : "2ec0121bc9dd5235afb2b7de48ffac3df012ff48",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15691",
       "triggerID" : "2c101e777cbdd1a2ec055ec62901be25d24da2f9",
       "triggerType" : "PUSH"
     }, {
       "hash" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "e50463182d1e1a09cb7d657d515fbbef60bdb668",
       "triggerType" : "PUSH"
     }, {
       "hash" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713",
       "triggerID" : "cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5",
       "triggerType" : "PUSH"
     }, {
       "hash" : "f6a937c54000be142978fc15aab23bb8fb33f701",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15750",
       "triggerID" : "f6a937c54000be142978fc15aab23bb8fb33f701",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * e50463182d1e1a09cb7d657d515fbbef60bdb668 UNKNOWN
   * cd29b7245f0c2db5c087355e6cc1f2e7d537d8e5 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15713) 
   * f6a937c54000be142978fc15aab23bb8fb33f701 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15750) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhuzhurk commented on a change in pull request #15314: [FLINK-21332][runtime] Optimize releasing result partitions in RegionPartitionReleaseStrategy

Posted by GitBox <gi...@apache.org>.
zhuzhurk commented on a change in pull request #15314:
URL: https://github.com/apache/flink/pull/15314#discussion_r603761487



##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewMaintainerTest.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.TestingSchedulingPipelinedRegion;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link ConsumerRegionGroupExecutionView} and {@link
+ * ConsumerRegionGroupExecutionViewMaintainer}.
+ */
+public class ConsumerRegionGroupExecutionViewMaintainerTest extends TestLogger {
+
+    TestingSchedulingExecutionVertex producer;

Review comment:
       these fields should be private

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/RegionPartitionReleaseStrategy.java
##########
@@ -101,27 +128,24 @@ private PipelinedRegionExecutionView getPipelinedRegionExecutionViewForVertex(
     }
 
     private List<IntermediateResultPartitionID> filterReleasablePartitions(
-            final Iterable<? extends SchedulingResultPartition> schedulingResultPartitions) {
-        return IterableUtils.toStream(schedulingResultPartitions)
-                .distinct()
-                .map(SchedulingResultPartition::getId)
-                .filter(this::areConsumerRegionsFinished)
-                .collect(Collectors.toList());
-    }
-
-    private boolean areConsumerRegionsFinished(
-            final IntermediateResultPartitionID resultPartitionId) {
-        final SchedulingResultPartition resultPartition =
-                schedulingTopology.getResultPartition(resultPartitionId);
-        return IterableUtils.toStream(resultPartition.getConsumers())
-                .map(SchedulingExecutionVertex::getId)
-                .allMatch(this::isRegionOfVertexFinished);
-    }
+            final Iterable<ConsumedPartitionGroup> consumedPartitionGroups) {
+
+        final List<IntermediateResultPartitionID> filteredPartitions = new ArrayList<>();

Review comment:
       nit: `filteredPartitions` -> `releasablePartitions`

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/partitionrelease/ConsumerRegionGroupExecutionViewMaintainer.java
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License
+ */
+
+package org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease;
+
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingPipelinedRegion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Track the finished progress of {@link ConsumerRegionGroupExecutionView}s.
+ *
+ * <p>NOTE: It only contains {@link SchedulingPipelinedRegion}s that have {@link
+ * ConsumedPartitionGroup}s.
+ */
+class ConsumerRegionGroupExecutionViewMaintainer {
+
+    /**
+     * The list of {@link ConsumerRegionGroupExecutionView}s that a SchedulingPipelinedRegion
+     * belongs to.
+     */
+    private final Map<SchedulingPipelinedRegion, List<ConsumerRegionGroupExecutionView>>

Review comment:
       A `Set` is conceptually better than a `List` here




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org