[FLINK-15375][core] Introduce MemorySize#toHumanReadableString.

flink-core/src/main/java/org/apache/flink/configuration/MemorySize.java

@@ -22,6 +22,7 @@
 
 import java.math.BigDecimal;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Optional;
@@ -54,6 +55,13 @@ public class MemorySize implements java.io.Serializable, Comparable<MemorySize>
 
  public static final MemorySize MAX_VALUE = new MemorySize(Long.MAX_VALUE);
 
+ private static final List<MemoryUnit> ORDERED_UNITS = Arrays.asList(
+ BYTES,
+ KILO_BYTES,
+ MEGA_BYTES,
+ GIGA_BYTES,
+ TERA_BYTES);
+
  // ------------------------------------------------------------------------
 
  /** The memory size, in bytes. */
@@ -62,6 +70,9 @@ public class MemorySize implements java.io.Serializable, Comparable<MemorySize>
  /** The memorized value returned by toString(). */
  private transient String stringified;
 
+ /** The memorized value returned by toHumanReadableString(). */
+ private transient String humanReadableStr;
+
  /**
  * Constructs a new MemorySize.
  *
@@ -132,23 +143,16 @@ public String toString() {
  }
 
  private String formatToString() {
- List<MemoryUnit> orderedUnits = Arrays.asList(
- BYTES,
- KILO_BYTES,
- MEGA_BYTES,
- GIGA_BYTES,
- TERA_BYTES);
-
- MemoryUnit highestIntegerUnit = IntStream.range(0, orderedUnits.size())
+ MemoryUnit highestIntegerUnit = IntStream.range(0, ORDERED_UNITS.size())
  .sequential()
- .filter(idx -> bytes % orderedUnits.get(idx).getMultiplier() != 0)
+ .filter(idx -> bytes % ORDERED_UNITS.get(idx).getMultiplier() != 0)
  .boxed()
  .findFirst()
  .map(idx -> {
  if (idx == 0) {
- return orderedUnits.get(0);
+ return ORDERED_UNITS.get(0);
  } else {
- return orderedUnits.get(idx - 1);
+ return ORDERED_UNITS.get(idx - 1);
  }
  }).orElse(BYTES);
 
@@ -158,6 +162,38 @@ private String formatToString() {
  highestIntegerUnit.getUnits()[1]);
  }
 
+ public String toHumanReadableString() {
+ if (humanReadableStr == null) {
+ humanReadableStr = formatToHumanReadableString();
+ }
+
+ return humanReadableStr;
+ }
+
+ private String formatToHumanReadableString() {
+ MemoryUnit highestUnit = IntStream.range(0, ORDERED_UNITS.size())
+ .sequential()
+ .filter(idx -> bytes > ORDERED_UNITS.get(idx).getMultiplier())
+ .boxed()
+ .max(Comparator.naturalOrder())
+ .map(ORDERED_UNITS::get)
+ .orElse(BYTES);
+
+ if (highestUnit == BYTES) {
+ return String.format(
+ "%d %s",
+ bytes,
+ BYTES.getUnits()[1]);
+ } else {
+ double approximate = 1.0 * bytes / highestUnit.getMultiplier();
+ return String.format(
+ "%.3f%s (%d bytes)",
+ approximate,
+ highestUnit.getUnits()[1],
+ bytes);
+ }
+ }
+
  @Override
  public int compareTo(MemorySize that) {
  return Long.compare(this.bytes, that.bytes);

flink-core/src/test/java/org/apache/flink/configuration/MemorySizeTest.java

@@ -233,4 +233,17 @@ public void testDivideByNegativeLong() {
  final MemorySize memory = new MemorySize(100L);
  memory.divide(-23L);
  }
+
+ @Test
+ public void testToHumanReadableString() {
+ assertThat(new MemorySize(0L).toHumanReadableString(), is("0 bytes"));
+ assertThat(new MemorySize(1L).toHumanReadableString(), is("1 bytes"));
+ assertThat(new MemorySize(1024L).toHumanReadableString(), is("1024 bytes"));
+ assertThat(new MemorySize(1025L).toHumanReadableString(), is("1.001kb (1025 bytes)"));
+ assertThat(new MemorySize(1536L).toHumanReadableString(), is("1.500kb (1536 bytes)"));
+ assertThat(new MemorySize(1_000_000L).toHumanReadableString(), is("976.563kb (1000000 bytes)"));
+ assertThat(new MemorySize(1_000_000_000L).toHumanReadableString(), is("953.674mb (1000000000 bytes)"));
+ assertThat(new MemorySize(1_000_000_000_000L).toHumanReadableString(), is("931.323gb (1000000000000 bytes)"));
+ assertThat(new MemorySize(1_000_000_000_000_000L).toHumanReadableString(), is("909.495tb (1000000000000000 bytes)"));
+ }
 }