[FLINK-31510][yarn] Use getMemorySize instead of getMemory.

[slfan1989]

What is the purpose of the change

JIRA: FLINK-31510. Use getMemorySize instead of getMemory

InYARN-4844, such a situation is described. We use int to represent memory (Unit MB). If a machine has 210GB of memory, the conversion to MB is 215040MB. If we have 10K machines, then 215040*10000=2150400000 > 2147483647 , accumulating memory will return a negative number.

In YARN-4844, use getMemorySize instead of getMemory, use Long type to represent memory instead of using Int type to represent memory.

Resource#getMemory

/**
   * This method is DEPRECATED:
   * Use {@link Resource#getMemorySize()} instead
   *
   * Get <em>memory</em> of the resource. Note - while memory has
   * never had a unit specified, all YARN configurations have specified memory
   * in MB. The assumption has been that the daemons and applications are always
   * using the same units. With the introduction of the ResourceInformation
   * class we have support for units - so this function will continue to return
   * memory but in the units of MB
   *
   * @return <em>memory</em>(in MB) of the resource
   */
  @Public
  @Deprecated
  public abstract int getMemory();

Resource#getMemorySize

/**
   * Get <em>memory</em> of the resource. Note - while memory has
   * never had a unit specified, all YARN configurations have specified memory
   * in MB. The assumption has been that the daemons and applications are always
   * using the same units. With the introduction of the ResourceInformation
   * class we have support for units - so this function will continue to return
   * memory but in the units of MB
   *
   * @return <em>memory</em> of the resource
   */
  @Public
  @Stable
  public long getMemorySize();

There is such a code in Flink to accumulate the memory returned by NM. This code may return a negative number.

YarnClusterDescriptor#getCurrentFreeClusterResources

int totalMemory = 0;
int totalCores = 0;
for (NodeReport rep : nodes) {
    final Resource res = rep.getCapability();
    totalMemory += res.getMemory();
    totalCores += res.getVirtualCores();
    ps.format(format, "NodeID", rep.getNodeId());
    ps.format(format, "Memory", getDisplayMemory(res.getMemory()));
    ps.format(format, "vCores", res.getVirtualCores());
    ps.format(format, "HealthReport", rep.getHealthReport());
    ps.format(format, "Containers", rep.getNumContainers());
    ps.println("+---------------------------------------+");
}

Brief change log

Use getMemorySize instead of getMemory.

Verifying this change

Manually verified.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? no

[flinkbot]

CI report:

• c68d1fe Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[chucheng92]

LGTM.(non-binding)

[chucheng92]

hi, i think here memory unit is MB. so we can use int, why must long ?

[reswqa]

Thanks @slfan1989 for creating this.

How did you manually verify this issue, could you describe it in detail?

By the way, would you mind resolving conflict in YarnClusterDescriptor?

[slfan1989]

Thanks @slfan1989 for creating this.

How did you manually verify this issue, could you describe it in detail?

By the way, would you mind resolving conflict in YarnClusterDescriptor?

@reswqa Thank you very much for your suggestion, I will modify the information as soon as possible

[slfan1989]

@reswqa Can you help review this PR again? Thank you very much! I explained the reason for possible negative numbers.

[reswqa]

Tanks @slfan1989 for the update, I only have one question.

[reswqa]

Thanks @slfan1989 for the update, would you mind squashing all to only one commit with the message like:

[FLINK-31510][yarn] Replace deprecated getMemory by getMemorySize

[slfan1989]

Thanks @slfan1989 for the update, would you mind squashing all to only one commit with the message like:

@reswqa Thank you for your help to review the code! I will rebase and submit.

[reswqa]

Thanks @slfan1989, LGTM.