zentol commented on code in PR #22883:
URL: https://github.com/apache/flink/pull/22883#discussion_r1245368209
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java:
##########
@@ -78,31 +80,54 @@ public ResourceCounter calculateRequiredSlots(
return ResourceCounter.withResource(ResourceProfile.UNKNOWN,
numTotalRequiredSlots);
}
- private static Map<SlotSharingGroupId, Integer>
getMaxParallelismForSlotSharingGroups(
- Iterable<JobInformation.VertexInformation> vertices) {
- final Map<SlotSharingGroupId, Integer>
maxParallelismForSlotSharingGroups = new HashMap<>();
+ private static <T> Map<SlotSharingGroupId, T> getPerSlotSharingGroups(
+ Iterable<JobInformation.VertexInformation> vertices,
+ Function<JobInformation.VertexInformation, T> mapper,
+ BiFunction<T, T, T> reducer) {
+ final Map<SlotSharingGroupId, T> extractedPerSlotSharingGroups = new
HashMap<>();
for (JobInformation.VertexInformation vertex : vertices) {
- maxParallelismForSlotSharingGroups.compute(
+ extractedPerSlotSharingGroups.compute(
vertex.getSlotSharingGroup().getSlotSharingGroupId(),
- (slotSharingGroupId, currentMaxParallelism) ->
- currentMaxParallelism == null
- ? vertex.getParallelism()
- : Math.max(currentMaxParallelism,
vertex.getParallelism()));
+ (slotSharingGroupId, currentData) ->
+ currentData == null
+ ? mapper.apply(vertex)
+ : reducer.apply(currentData,
mapper.apply(vertex)));
}
- return maxParallelismForSlotSharingGroups;
+ return extractedPerSlotSharingGroups;
+ }
+
+ private static Map<SlotSharingGroupId, Integer>
getMinParallelismPerSlotSharingGroup(
+ Iterable<JobInformation.VertexInformation> vertices) {
+ return getPerSlotSharingGroups(
+ vertices, JobInformation.VertexInformation::getMinParallelism,
Math::min);
+ }
+
+ private static Map<SlotSharingGroupId, Integer>
getMaxParallelismForSlotSharingGroups(
+ Iterable<JobInformation.VertexInformation> vertices) {
+ return getPerSlotSharingGroups(
+ vertices, JobInformation.VertexInformation::getParallelism,
Math::max);
Review Comment:
nope, but the confusion is understandable. Max parallelism is an overloaded
term, both referring to the max parallelism a job can ever run (== the number
of key groups), and the upper bound parallelism that the job can run at.
Outside of validation purposes the actual max parallelism isn't relevant for
scaling.
We should rename things to explicitly refer to lower/upper parallelism bound.
--
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]
