Adapt to Flink v1.11 memory configuration

[alberto-quesada-scopely]

The memory setup has changed a lot with the 1.10 release for TaskManagers and with the 1.11 release for JobManagers.

The options set by the operator (jobmanager.heap.size and taskmanager.heap.size) are now deprecated. They can still be used in order to maintain backwards compatibility but they will be interpreted as a different option. Full migration guide can be found here.

With this change, when the version of Flink that we are deploying is lower than 1.11 nothing will change: memory options will be computed as it was before. When the Flink version is 1.11 or above, we will set jobmanager.memory.process.size and taskmanager.memory.process.size using the user defined requested memory.

If no Flink version is specified or we set an incorrect version (non semantic versioning model for example), default configuration will be set.

A new dependency has been added: go-version. It helps parsing semantic versioning and comparing between versions.

[kelly-sm]

Shouldn't this logic be applied for TM in version >=1.10? What's the reasoning behind only adding it for >= 1.11?

Without this logic in place in 1.10 - I have to manually configure taskmanager.memory.process.size to match with the computed heap size or else I get validation errors.

[alberto-quesada-scopely]

Shouldn't this logic be applied for TM in version >=1.10? What's the reasoning behind only adding it for >= 1.11?

Without this logic in place in 1.10 - I have to manually configure taskmanager.memory.process.size to match with the computed heap size or else I get validation errors.

Only half of this configuration is valid for the version 1.10. In the version 1.11 they added the same behaviour to the JM for consistency. We could change this so when version is >= 1.10, we would set taskmanager.memory.process.size and when version is >= 1.11 we add jobmanager.memory.process.size as well.

[kelly-sm]

Shouldn't this logic be applied for TM in version >=1.10? What's the reasoning behind only adding it for >= 1.11? Without this logic in place in 1.10 - I have to manually configure taskmanager.memory.process.size to match with the computed heap size or else I get validation errors.

Only half of this configuration is valid for the version 1.10. In the version 1.11 they added the same behaviour to the JM for consistency. We could change this so when version is >= 1.10, we would set taskmanager.memory.process.size and when version is >= 1.11 we add jobmanager.memory.process.size as well.

That makes more sense to me, just wanted to confirm whether I was missing something/misread the Flink docs.

[mwylde]

Hey @alberto-quesada-scopely — have you had a chance to look at my comment? At Lyft we're also looking to roll out Flink 1.11 so would be great to get this change in.

[alberto-quesada-scopely]

Hey @alberto-quesada-scopely — have you had a chance to look at my comment? At Lyft we're also looking to roll out Flink 1.11 so would be great to get this change in.

Hey! I've been on holidays but I will try to adapt it asap ;)

[alberto-quesada-scopely]

Hi @mwylde I just pushed some changes, but I'm not super happy with the name offProcessMemoryFraction either... is hard to come up with something that works fine for both off heap and off process... Other than that, I also added support for process memory in task managers for version 1.10.

[mwylde]

Looks like there's a merge conflict — can you rebase on top of master?

[alberto-quesada-scopely]

Looks like there's a merge conflict — can you rebase on top of master?

Done! Thanks!