wanglijie95 commented on code in PR #22674:
URL: https://github.com/apache/flink/pull/22674#discussion_r1244774850
##########
flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java:
##########
@@ -221,14 +224,17 @@ private ShuffleDescriptorAndIndex[]
computeConsumedPartitionShuffleDescriptors(
}
private MaybeOffloaded<ShuffleDescriptorAndIndex[]>
serializeAndTryOffloadShuffleDescriptor(
- ShuffleDescriptorAndIndex[] shuffleDescriptors) throws IOException
{
+ ShuffleDescriptorAndIndex[] shuffleDescriptors, int numConsumer)
throws IOException {
Review Comment:
Can we abstract the logic here to a inline class
`ShuffleDescriptorSerializer`, and provide 2 method:
`serializeShuffleDescriptor` and `serializeAndTryOffloadShuffleDescriptor`. And
then pass the serializer to `CachedShuffleDescriptors`, the
`CachedShuffleDescriptors` will decide to call which method according to the
`shouldOffload`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/deployment/CachedShuffleDescriptors.java:
##########
@@ -54,10 +54,14 @@ public class CachedShuffleDescriptors {
/** Stores the mapping of resultPartitionId to index subscripts in
consumed partition group. */
private final Map<IntermediateResultPartitionID, Integer>
resultPartitionIdToIndex;
+ /** The number of consumers for {@link ConsumedPartitionGroup}. */
+ private final int numConsumers;
+
public CachedShuffleDescriptors(
ConsumedPartitionGroup consumedPartitionGroup,
ShuffleDescriptorAndIndex[] shuffleDescriptors) {
Review Comment:
How about pass the `shouldOffload` in the ctor of `CachedShuffleDescriptors`
? We can judge by
```
consumedPartitionGroup.size() * consumedPartitionGroup.getNumConsumers() >=
offloadShuffleDescriptorsThreshold
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
