zhuzhurk commented on a change in pull request #10179: [FLINK-14734][runtime] Add a ResourceSpec in SlotSharingGroup to describe its overall resources URL: https://github.com/apache/flink/pull/10179#discussion_r348313399
########## File path: flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java ########## @@ -149,6 +150,39 @@ public ResourceSpec merge(final ResourceSpec other) { return target; } + /** + * Subtracts another resource spec from this one. + * + * @param other The other resource spec to subtract. + * @return The subtracted resource spec. + */ + public ResourceSpec subtract(final ResourceSpec other) { + checkNotNull(other, "Cannot subtract null resources"); + + if (this.equals(UNKNOWN) || other.equals(UNKNOWN)) { + return UNKNOWN; + } + + checkArgument(other.lessThanOrEqual(this), "Cannot subtract a larger ResourceSpec from this one."); + + final ResourceSpec target = new ResourceSpec( + this.cpuCores.subtract(other.cpuCores), + this.taskHeapMemory.subtract(other.taskHeapMemory), + this.taskOffHeapMemory.subtract(other.taskOffHeapMemory), + this.onHeapManagedMemory.subtract(other.onHeapManagedMemory), + this.offHeapManagedMemory.subtract(other.offHeapManagedMemory)); + + target.extendedResources.putAll(extendedResources); + + for (Resource resource : other.extendedResources.values()) { + target.extendedResources.merge(resource.getName(), resource, (v1, v2) -> { + final Resource subtracted = v1.subtract(v2); + return subtracted.getValue().compareTo(BigDecimal.ZERO) == 0 ? null : subtracted; Review comment: FLINK-14869 is opened to track this issue. ---------------------------------------------------------------- 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: us...@infra.apache.org With regards, Apache Git Services