[FLINK-14086][core] Add OperatingArchitecture Enum

flink-core/src/main/java/org/apache/flink/util/MemoryArchitecture.java → flink-core/src/main/java/org/apache/flink/util/ProcessorArchitecture.java

@@ -26,10 +26,16 @@
  * Note that this might be different than the actual operating system's architecture, for example
  * when installing a 32 bit JRE in a 64 bit OS.
  */
-public enum MemoryArchitecture {
+public enum ProcessorArchitecture {
+ /**
+ * X86 platform.
+ */
+ X86,
 
- // constants here start with an underscore because Java identifier cannot start with a
- // numeric character and alternatives like 'BIT_64' are not as readable
+ /**
+ * arm platform.
+ */
+ ARM,
 
  /**
  * 32 bit memory address size.
@@ -46,18 +52,25 @@ public enum MemoryArchitecture {
  */
  UNKNOWN;
 
- // ------------------------------------------------------------------------
 
- private static final MemoryArchitecture current = getInternal();
+ private static final ProcessorArchitecture arch = readArchFromSystemProperties();
+ private static final ProcessorArchitecture size = readSizeFromSystemProperties();
 
- /**
- * Gets the processor architecture of this process.
- */
- public static MemoryArchitecture get() {
- return current;
+ private static ProcessorArchitecture readArchFromSystemProperties() {
+ final List<String> namesX86 = Arrays.asList("amd64", "x86_64", "x86", "i386", "i486", "i586", "i686");
+ final List<String> namesArm = Arrays.asList("arm", "aarch64");
+ final String arch = System.getProperty("os.arch");
+
+ if (namesX86.contains(arch)) {
+ return X86;
+ } else if (namesArm.contains(arch)) {
+ return ARM;
+ } else {
+ return UNKNOWN;
+ }
  }
 
- private static MemoryArchitecture getInternal() {
+ private static ProcessorArchitecture readSizeFromSystemProperties() {
  // putting these into the method to avoid having objects on the heap that are not needed
  // any more after initialization
  final List<String> names64bit = Arrays.asList("amd64", "x86_64", "aarch64");
@@ -66,12 +79,19 @@ private static MemoryArchitecture getInternal() {
 
  if (names64bit.contains(arch)) {
  return _64_BIT;
- }
- else if (names32bit.contains(arch)) {
+ } else if (names32bit.contains(arch)) {
  return _32_BIT;
- }
- else {
+ } else {
  return UNKNOWN;
  }
  }
+
+ public static ProcessorArchitecture getCurrentOperatingSystemArch() {
+ return arch;
+ }
+
+ public static ProcessorArchitecture getCurrentOperatingSystemSize() {
+ return size;
+ }
+
 }

flink-core/src/test/java/org/apache/flink/util/MemoryArchitectureTest.java → flink-core/src/test/java/org/apache/flink/util/ProcessorArchitectureTest.java

@@ -23,14 +23,16 @@
 import static org.junit.Assert.assertNotEquals;
 
 /**
- * Tests for the {@link MemoryArchitecture}.
+ * Tests for the {@link ProcessorArchitecture}.
  */
-public class MemoryArchitectureTest {
+public class ProcessorArchitectureTest {
 
  @Test
  public void testArchitectureNotUnknown() {
- final MemoryArchitecture arch = MemoryArchitecture.get();
+ final ProcessorArchitecture arch = ProcessorArchitecture.getCurrentOperatingSystemArch();
+ final ProcessorArchitecture size = ProcessorArchitecture.getCurrentOperatingSystemSize();
 
- assertNotEquals(MemoryArchitecture.UNKNOWN, arch);
+ assertNotEquals(ProcessorArchitecture.UNKNOWN, arch);
+ assertNotEquals(ProcessorArchitecture.UNKNOWN, size);
  }
 }

flink-end-to-end-tests/flink-metrics-reporter-prometheus-test/src/test/java/org/apache/flink/metrics/prometheus/tests/PrometheusReporterEndToEndITCase.java

@@ -28,6 +28,7 @@
 import org.apache.flink.tests.util.categories.TravisGroup1;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.OperatingSystem;
+import org.apache.flink.util.ProcessorArchitecture;
 import org.apache.flink.util.TestLogger;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
@@ -68,17 +69,52 @@ public class PrometheusReporterEndToEndITCase extends TestLogger {
 
  static {
  final String base = "prometheus-" + PROMETHEUS_VERSION + '.';
+ final String os;
+ final String platform;
  switch (OperatingSystem.getCurrentOperatingSystem()) {
  case MAC_OS:
- PROMETHEUS_FILE_NAME = base + "darwin-amd64";
+ os = "darwin";
  break;
  case WINDOWS:
- PROMETHEUS_FILE_NAME = base + "windows-amd64";
+ os = "windows";
  break;
  default:
- PROMETHEUS_FILE_NAME = base + "linux-amd64";
+ os = "linux";
  break;
  }
+ switch (ProcessorArchitecture.getCurrentOperatingSystemArch()) {
+ case X86:
+ switch (ProcessorArchitecture.getCurrentOperatingSystemSize()) {
+ case _32_BIT:
+ platform = "386";
+ break;
+ case _64_BIT:
+ platform = "amd64";
+ break;
+ default:
+ platform = "Unknown";
+ break;
+ }
+ break;
+ case ARM:
+ switch (ProcessorArchitecture.getCurrentOperatingSystemSize()) {
+ case _32_BIT:
+ platform = "armv7";
+ break;
+ case _64_BIT:
+ platform = "arm64";
+ break;
+ default:
+ platform = "Unknown";
+ break;
+ }
+ break;
+ default:
+ platform = "Unknown";
+ break;
+ }
+
+ PROMETHEUS_FILE_NAME = base + os + "-" + platform;
  }
 
  private static final Pattern LOG_REPORTER_PORT_PATTERN = Pattern.compile(".*Started PrometheusReporter HTTP server on port ([0-9]+).*");

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java

@@ -28,7 +28,7 @@
 import org.apache.flink.runtime.io.network.buffer.BufferPoolOwner;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.MemoryArchitecture;
+import org.apache.flink.util.ProcessorArchitecture;
 import org.apache.flink.util.function.FunctionWithException;
 
 import org.slf4j.Logger;
@@ -220,7 +220,7 @@ FunctionWithException<BufferPoolOwner, BufferPool, IOException> createBufferPool
  }
 
  static BoundedBlockingSubpartitionType getBoundedBlockingType() {
- switch (MemoryArchitecture.get()) {
+ switch (ProcessorArchitecture.getCurrentOperatingSystemSize()) {
  case _64_BIT:
  return BoundedBlockingSubpartitionType.FILE_MMAP;
  case _32_BIT: