[FLINK-12647][network] Add feature flag to disable release of consumed blocking partitions

[zentol]

What is the purpose of the change

Based on #8608; only the last 2 commits are relevant.

The first commit introduces the concept of externally managed partitions, i.e. partitions that are not automatically released by the ShuffleEnvironment on consumption, but instead when instructed to do so by the TaskManager, or more generally as part of the partition lifecycle management. On it's own this commit does not change any behavior, as no part of the system makes use of this flag.
This is primarily a preparatory step for the subsequent commit and future partition lifecycle management.

The second commit introduces a dev feature flag to enable/disable the automatic release of consumed blocking result partitions by the shuffle environment.

If the flag is enabled, which is the default, then the system does not behave differently to how it does currently.
Once the partition lifecycle is fully introduced we will either switch the default to false, or remove the flag entirely.

[flinkbot]

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.

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 for a full explanation of the review process.

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

Bot commands

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

[zhijiangW]

Thanks for opening this PR @zentol!

It looks general good to me and only left some small comments.

[zhijiangW]

Thanks for the updates @zentol.
LGTM!

[zentol]

@zhijiangW could have another look at the latest commit? Addresses an issue raised in #8687 . #8687 (comment)

[zhijiangW]

nit: I guess we do not need set partition type and number of subpartitions, and the default would make sense.

[zentol]

The partition being blocking is important for the test so I'd like to not omit it.

[zhijiangW]

Actually I thought the property of isExternallyManaged should control whether it could be consumed multiple times not related to ResultPartitionType. But in reality the pipelined type could not support this atm because of the checkState(readView == null while PipelinedSubpartition#createReadView. So we can keep the current way for setting the blocking type.

[zhijiangW]

Yes, I have reviewed the last commit and thought you might create separate ticket for ref-count before. :)

[zentol]

this isn't sufficient; the partition is already removed from registeredPartitions above, leading to wrong behavior of getUnreleasedPartitions

[zhijiangW]

Yes, I also noticed this just now, the state is inconsistent here if removed but not released.

[zhijiangW]

if isManagedExternally, we should not call partitionManager.onConsumedPartition(this) after one sub partition is consumed.

[zentol]

It should still be safe to do so (because the RPM basically ignores the call), but yeah this shows that I haven't thought this through properly.

[zentol]

Could we call pin() in createSubpartitionView and decrement the refcount as usual in onConsumedPartition?

[zhijiangW]

The pin would increase the counter as the number of subpartitions, but createSubpartitionView only indicates for one of the subpartitions. I need think through this issue again.
Maybe we could focus on the pin issue in a separate ticket and this PR only introduces the flag but not changed any current behaviors if you like.

[zentol]

I think I've found a way to do it; pin() will continue to increment the refCount by the number of subpartitions, but createSubpartitionsView will increment it by 1 (if the partition is externally managed).

I'm not sure whether this is a good approach, and what plans people had on how this should be implemented, so I wouldn't oppose deferring this to a follow-up.

[zhijiangW]

TBH I do not like the current way because it seems a bit hacky.
Because during ResultPartition#onConsumedSubpartition, the counter would never be 0 for external case, so the ResultPartitionManager#onConsumedPartition would never be called, then the internal checking logic in ResultPartitionManager#onConsumedPartition seems redunctant.

[zhijiangW]

Maybe we could always pin as before, but migrate the checking logic from ResultPartitionManger to ResultPartition as I mentioned before. I mean ResultPartitionManager is not aware of the external issue, the tag in ResultPartitionManager#isReleaseExternallyManagedPartitionsOnConsumption could be merged with ResultPartition#isManagedExternally to generate a final tag in ResultPartition.

Then while ResultPartition#onConsumedSubpartition, it could check the final tag to decide whether to call ResultPartitionManager#onConsumedPartition. And the reference counter only works when the final tag is false.

[zhijiangW]

In conclusion:

• do not modify previous ResultPartitionManger, only add final tag in ResultPartition

• pin logic could keep same as before.

• the ref counter check in ResultPartition#createSubpartitionView only works when tag false

• the ref counter check in ResultPartition#onConsumedSubpartition only works when tag false

[zhijiangW]

I have another option for this. ATM both ResultPartition and ResultPartitionManager need check the conditions of releasing partitions consumed, then the logic seems difficult to trace.
We could make ResultPartitionManager not aware of this. The isReleaseExternallyManagedPartitionsOnConsumption could be merged with partition.isManagedExternally() to generate one final boolean tag in ResultPartition. Then the ResultPartition checks this tag to decide whether to call partitionManager.onConsumedPartition while one sub partition is consumed. The logic in ResultPartitionManager would be simple:

• Based on consumption release: The behavior is the same as before after calling ResultPartitionManger#onConsumedPartition, remove from map and release directly.

• Based on external release: via ResultPartitionManager#releasePartition.

And we only have one boolean tag in ResultPartition to handle the external check. WDYT?
This suggestion is not relevant with pin issue.

[zentol]

... I never thought about combining isManagedExternally and the configuration flag. That's a great idea.

[zentol]

Implemented.

[tillrohrmann]

Thanks for opening this PR @zentol and the review @zhijiangW. I had some more comments. In particular, I would be interested whether we want to introduce a ReleaseOnConsumptionResultPartition which encapsulates the release logic.

[zentol]

@tillrohrmann I implemented both of your suggestions.

I like the renaming of externallyManaged to releasedOnConsumption; it makes things a lot easier to understand.

As for the ResultPartition sub-class, the empty pin() method in the base-class is a bit odd, but otherwise I like it.

[tillrohrmann]

Thanks for addressing my comments @zentol. LGTM. +1 for merging once Travis gives green light.

[zhijiangW]

Actually I am not sure why it needs while loop here before.

[gaoyunhaii]

Follow Zhijiang's thought, I think we can now remove the pin method from the ResultPartition, since it is only the implementation detail of the ReleaseOnConsumptionResultPartiton. Then we can move the initialization of the pending reference to the Constructor to ensure no concurrent access to the variable, then we can use pendingReferences.set(subpartitions.length) to initialize the variable.

[zhijiangW]

I created FLINK-12842 and FLINK-12843 for the ref-counter issues which could be solved separately after this PR merged, because these issues already exist before and are not in the scope of this PR.

[zentol]

I agree that we can resolve these in follow-ups.

[zhijiangW]

Thanks for the updates @zentol and the reviews @tillrohrmann .
I like the current way and it looks pretty good now. I only left several nit comments.