zhijiangW commented on a change in pull request #8789: [FLINK-12890] Add
partition lifecycle related Shuffle API
URL: https://github.com/apache/flink/pull/8789#discussion_r296586155
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/ShuffleMaster.java
##########
@@ -41,4 +44,17 @@
CompletableFuture<T> registerPartitionWithProducer(
PartitionDescriptor partitionDescriptor,
ProducerDescriptor producerDescriptor);
+
+ /**
+ * Release any external resources occupied by the given partition.
+ *
+ * <p>This call triggers release of any resources which are occupied by
the given partition in the external systems
+ * outside of the producer executor. This is mostly relevant for the
batch jobs and blocking result partitions
+ * for which {@link
ResultPartitionDeploymentDescriptor#isReleasedOnConsumption()} returns {@code
false}.
+ * The producer local resources are managed by {@link
ShuffleDescriptor#hasLocalResources()} and
+ * {@link ShuffleEnvironment#releasePartitions(Collection)}.
+ *
+ * @param shuffleDescriptor shuffle descriptor of the result partition
to release externally.
+ */
+ void releasePartitionExternally(T shuffleDescriptor);
Review comment:
Thanks for the above detail explanations and I could understand the
functions/semantics we want to provide. I think we provide three ways for
releasing partitions atm as below table:
Partition release | Interface dependencies
-- | --
1.release once consumed | PD#releaseOnConsumption
2.TM internal shuffle RPC-based | JM->TM RPC,
ShuffleEnvironment#releasePartitionsLocally, SD#storesLocalResourcesOn
3.external shuffle RPC-base | ShuffleMaster#releasePartitionsExternally,
SD#storesLocalResourcesOn
My previous thought was considering whether the interface and proposed
methods are clear enough for the third-party implementation of
`ShuffleManager`. We could see from above table that there are four interface
methods involved in partition release (PD, SD, ShuffleMaster,
ShuffleEnvironment) and some methods are coupled with each other.
E.g. we assume if `SD#storesLocalResourcesOn` is true,
`ShuffleEnvironment#releasePartitionsLocally` should be used for releasing
partition, otherwise `ShuffleMaster#releasePartitionsExternally` should be
used. Actually we both agree that for the above case 2 in table, we could also
rely on `ShuffleMaster#releasePartitionsExternally` to release partition if we
want, and only we rely on JM/TM RPC as shortcut default implementation.
But if somebody wants to implement another `ShuffleManager` to replace the
current one for case 2, he might be confused of why
`ShuffleMaster#releasePartitionsExternally` was not suggested using and what is
`Externally` indicating for. If the interface was only defined as
`ShuffleMaster#releasePartitions`, it does not need to explain the differences
between externally and locally which might bring confusing sometimes. It only
indicates there triggers a release call from `ShuffleMaster`, the third-party
could decide whether to implement it or empty via other ways.
Also for `ShuffleEnvironment#releasePartition` it only indicates to trigger
a partition release call from `ShuffleEnvironment`, no need to limit/emphasize
it for a certain specific implementation, although the external case 3 might
never call it in implementation.
Actually from above table we could see we need two boolean tag infos atm.
One is for describing the partition release mechanism, and the other is for
indicating TM internal shuffle or external shuffle. I think these two infos
might be both tagged in `ShuffleDescriptor` in future. Because we assume the
pipelined partition could be consumed only once limited by current
implementation, so we tag it in PD#releaseOnConsumption. But whether a
partition could be consumed multiple times is also up to the implementation of
`ShuffleEnvironment#createResultPartitionWriters`, we could also implement a
pipelined partition which could be consumed multi times. So the SD could
consider the partition type with internal implementation to give a final tag.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
