[FLINK-40467] Randomise application result store paths - #1199
Conversation
james-kan-shopify
left a comment
james-kan-shopify
left a comment
There was a problem hiding this comment.
thanks for opening this up! happen to encounter this and attempting to patch the same thing here.
| } | ||
|
|
||
| private static void setRandomApplicationResultStorePath(Configuration effectiveConfig) { | ||
| if (effectiveConfig.contains(HighAvailabilityOptions.HA_STORAGE_PATH)) { |
There was a problem hiding this comment.
Hi! I've encountered the same bug, and have been working in a fork regarding this. Thanks for opening up a PR here!
Suggesting to consider also setting application-result-store.delete-on-commit=false, mirroringsetRandomJobResultStorePath.
maybeSubmitApplicationInApplicationMode skips submission on a clean result entry, but per its javadoc that lookup only works when DELETE_ON_COMMIT=false. Since the operator sets shutdown-on-application-finish=false, a terminal app's JM lives on until the shutdown TTL, and a restart in that window re-bootstraps it: main() re-runs, and multi-job apps fail outright. FLINK-27569 one level up.
There was a problem hiding this comment.
I do think that's worth fixing, but IMO it's a separate enough issue that it'd be better in a separate pull request (maybe a separate Jira bug issue too)
There was a problem hiding this comment.
If we need delete on commit false here as well same way as the job result store then we should fix it in the same PR / issue
There was a problem hiding this comment.
@gyfora I'm happy to do that if you're okay with the approach in the code here
James fairly highlighted the resource leak implication of the approach I took on the dev@ list where I said:
re: resource leak
Yeah, I agree that what I'm proposing in FLINK-40467 doubles down on the deal made in FLINK-27573, and maybe that's not a good thing. I used a random UUID because that's what I saw setRandomJobResultStorePath do, but in hindsight making directories anonymous makes the admin cleanup task harder as you can't derive the deployment from a directory name to know what is safe to remove. Perhaps the operator should record the path it generates in status - this would let us automate cleaning up the path for previous launches?
if you want to discuss that aspect first, I'll hold off on adding to the PR. But if you're okay with the approach, I'll add the delete on commit
There was a problem hiding this comment.
I think we have the exact same result leak already for the job result store logic. Not setting this config means that we have an inconsistent behaviour between different flink versions. We can consider how to avoid the resource leak as the next step but for now I think it makes sense to follow the currently established approach
Dennis-Mircea
left a comment
Dennis-Mircea
left a comment
There was a problem hiding this comment.
Thanks for the fix, @dalelane! I posted a detailed analysis on the dev thread (https://lists.apache.org/thread/tyw45k5ww3yzfmfkmhz241nm82swlrsm) arguing that the current issue, unlike FLINK-27569, is an ordering problem (the operator tears the JM down before Flink deletes the dirty application result) and can be solved without introducing a second permanent leak. I would suggest holding this PR until the direction is agreed.
|
Can you please rebase on main to pull in the e2e fix? |
Flink 2.3 gives every application-mode execution an
ApplicationResultStore, which defaults to
{high-availability.storage-path}/application-result-store/{ha.cluster-id}.
That path is consistent across deployments (and not included in HA
cleanup) so entries written by a terminated application outlive the
upgrade that replaced it.
SessionDispatcherLeaderProcess recovers dirty results unkeyed, so the
replacement cluster picks up the previous deployment's entry, and
Dispatcher#maybeSubmitApplicationInApplicationMode then deliberately
submits nothing when execution.shutdown-on-application-finish is false.
This results in healthy JobManagers with no jobs.
This commit makes the storage path unique on each cluster launch so
nothing stale can be recovered.
Signed-off-by: Dale Lane <dale.lane@uk.ibm.com>
job-result-store.delete-on-commit is currently set to false so the terminal result survives for applications with HA storage configured, to protect against terminated jobs inappropriately being restarted. This commit applies the application-mode equivalent of the same thing so that Job Manager restarts after an application has reached don't get resubmitted. This does result in a resource leak that needs manual cleanup, which is described in a docs update. Signed-off-by: Dale Lane <dale.lane@uk.ibm.com>
a0fe7fb to
251be41
Compare
4 participants
What is the purpose of the change
Flink 2.3 gives every application-mode execution an ApplicationResultStore, which defaults to
{high-availability.storage-path}/application-result-store/{ha.cluster-id}. That path is consistent across deployments (and not included in HA cleanup) so entries written by a terminated application outlive the upgrade that replaced it.SessionDispatcherLeaderProcess recovers dirty results unkeyed, so the replacement cluster picks up the previous deployment's entry, and Dispatcher#maybeSubmitApplicationInApplicationMode then deliberately submits nothing when execution.shutdown-on-application-finish is false. This results in healthy JobManagers with no jobs.
Brief change log
This commit makes the storage path unique on each cluster launch so nothing stale can be recovered.
Verifying this change
Added a unit test that recreates the scenario
Does this pull request potentially affect one of the following parts:
CustomResourceDescriptors: noDocumentation
Was generative AI tooling used to co-author this PR?