[hotfix][core] Introduce divide operation to Resource

flink-core/src/main/java/org/apache/flink/api/common/resources/Resource.java

@@ -22,6 +22,7 @@
 
 import java.io.Serializable;
 import java.math.BigDecimal;
+import java.math.RoundingMode;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -67,6 +68,14 @@ public Resource subtract(Resource other) {
  return create(value.subtract(other.value));
  }
 
+ public Resource divide(BigDecimal by) {
+ return create(value.divide(by, 16, RoundingMode.DOWN));
+ }
+
+ public Resource divide(int by) {
+ return divide(BigDecimal.valueOf(by));
+ }
+
  @Override
  public boolean equals(Object o) {
  if (this == o) {

flink-core/src/test/java/org/apache/flink/api/common/resources/ResourceTest.java

@@ -98,6 +98,34 @@ public void testSubtractErrorOnDifferentTypes() {
  v1.subtract(v2);
  }
 
+ @Test
+ public void testDivide() {
+ final Resource resource = new TestResource(0.04);
+ final BigDecimal by = BigDecimal.valueOf(0.1);
+ assertTestResourceValueEquals(0.4, resource.divide(by));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDivideNegative() {
+ final Resource resource = new TestResource(1.2);
+ final BigDecimal by = BigDecimal.valueOf(-0.5);
+ resource.divide(by);
+ }
+
+ @Test
+ public void testDivideInteger() {
+ final Resource resource = new TestResource(0.12);
+ final int by = 4;
+ assertTestResourceValueEquals(0.03, resource.divide(by));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDivideNegativeInteger() {
+ final Resource resource = new TestResource(1.2);
+ final int by = -5;
+ resource.divide(by);
+ }
+
  private static void assertTestResourceValueEquals(final double value, final Resource resource) {
  assertEquals(new TestResource(value), resource);
  }