[FLINK-11470][runtime] Pass configurations to filesystems when executing in LocalStreamEnvironment #17580

fapaul · 2021-10-27T14:29:10Z

What is the purpose of the change

Before this change, Flink's configuration where not passed to the
filesytems if used in LocalStreamEnvironment. This may lead to i.e.
missing credentials. This change initializes the filesytems through the
pluginloader which forwards the configuration as it is done when
running a real cluster.

Brief change log

Build the pluginloader in the Minicluster and initialize the filesystems with it.

Verifying this change

Added a test to verify that the configuration is available within the filesystem while running in the LocalStreamEnvironment

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? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2021-10-27T14:32:56Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit bbad41f (Wed Oct 27 14:32:56 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-10-27T14:33:03Z

CI report:

f8a65a0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flink-clients/src/test/java/org/apache/flink/client/ExecutionEnvironmentTest.java

AHeise

This mostly looks good to me. However, I'd like to get a second opinion as quite a few things are not clear enough to me.

AHeise · 2021-11-01T07:45:26Z

flink-clients/src/test/java/org/apache/flink/client/ExecutionEnvironmentTest.java

 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.DiscardingOutputFormat;


Commit + PR message should rather read: Initialize filesystems when run in LocalStreamEnvironment.
The issue is not that configuration was incorrectly passed to the fs but rather that plugins where not initialized at all.

The filesystems are initialized anyways but if we do not load them through the pluginloader the configuration is not passed to them

Hm, we should probably remove the fallback and fail hard to exactly find these kind of issues.

Btw same question as below: have we actually initialized plugins before your fix at all? I don't think so.

flink-clients/src/test/java/org/apache/flink/client/ExecutionEnvironmentTest.java

AHeise · 2021-11-01T07:53:59Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

+                final PluginManager pluginManager =
+                        PluginUtils.createPluginManagerFromRootFolder(configuration);
+                FileSystem.initialize(configuration, pluginManager);
+


Is MiniCluster always and only used for LocalExecutionEnvironment? I think the fix is legit but I'm wondering if the description is accurate. Are we solving an issue in the LocalExecutor or in MiniCluster?
I also don't quite understand why ClusterEntrypoint is not used at all in LocalExecutionEnvironment. @zentol could you please also have peek?

Is MiniCluster always and only used for LocalExecutionEnvironment?

No. It can also be used if the Test(Stream)Environments are explicitly set by the user.

The ClusterEntryPoint is used for actual clusters where the TMs run externally. The MiniCluster is the single-JVM equivalent.

AHeise · 2021-11-01T07:56:57Z

flink-clients/src/test/java/org/apache/flink/client/ExecutionEnvironmentTest.java

+        final StreamExecutionEnvironment environment = new LocalStreamEnvironment(config);
+        environment.getCheckpointConfig().setCheckpointStorage("testConfig:/" + tmp.newFolder());
+        environment.enableCheckpointing(100);
+        environment.fromElements(1, 2, 3, 4).map((MapFunction<Integer, Integer>) value -> value);


Is this cast needed?

It is not about the cast but rather the execution, without inserting the mapper the job is not executable.

flink-clients/src/test/java/org/apache/flink/client/ExecutionEnvironmentTest.java

flink-clients/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory

zentol

While I get the idea I'm very concerned about the singleton nature of the FileSystem class, and I don't know how to resolve that.

zentol · 2021-11-02T12:29:28Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

+                final PluginManager pluginManager =
+                        PluginUtils.createPluginManagerFromRootFolder(configuration);
+                FileSystem.initialize(configuration, pluginManager);
+


The ClusterEntryPoint is used for actual clusters where the TMs run externally. The MiniCluster is the single-JVM equivalent.

zentol · 2021-11-02T12:44:42Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java


            try {
+                final PluginManager pluginManager =
+                        PluginUtils.createPluginManagerFromRootFolder(configuration);


This pretty much doesn't do anything so I'd suggest to use FileSystem#initialize(Configuration).

The FileSystem#initialize(Configuration) method is marked as deprecated I did not want to add a usage again.

then you can also pass null for the PluginManager.

Why is null sufficient? I think the pluginManager is needed such that files in the plugin folder are loaded at all. That's the only way to read/write to S3 in local setup. I'm probably missing something here.

Filesystems can still be loaded through the FileSystemFactory service if they are on the classpath.
Plugins in local environments are a bit eh because there's a) no sane way to control the plugin directory (no, setting some internal env variable is not sane) and b) you must not add a dependency on said FS otherwise you lose all the benefits.

Yes, but users are trying out their Flink job locally and use a plugin folder in their working directories for s3 (that works by default).

Just from the last month: https://lists.apache.org/thread/kojvkx9ops9fmdkkcwsmpzkwrx94wcym https://lists.apache.org/thread/2dnlk7lc489ts20338cl3ll8gvfx6f21

zentol · 2021-11-02T12:46:26Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

            try {
+                final PluginManager pluginManager =
+                        PluginUtils.createPluginManagerFromRootFolder(configuration);
+                FileSystem.initialize(configuration, pluginManager);


This seems sketchy. Because FileSystem is partially a singleton this could have surprising effects on other tests if a FS is provided for one of the default schemes (e.g., a custom LocalFileSystem).

Would it be safer to move it to the LocalExecutor?

I don't think so because all tests using the MiniClusters implicitly go through the LocalEnvironment.

While it is true that users cannot set credentials via the configuration passed to the environment, they could call FileSystem#initialize() themselves.
Yes, it is surprising and not ergonomic and all that.

Alternatively we would need a way to reset the FileSystem (i.e., call initialize again with no configuration) after each test.

How about we add a test-only uninitialize? Wouldn't that be the cleanest option?

And who calls that? Every test that uses a filesystem? Every test that does something special? How do you enforce that?

Every test that calls initialize must call uninitialize. We can't enforce it but rather make sure that all existing tests are abiding it.

Tests might not explicitly call initialize; if they have a factory on the CP then it is loaded implicitly on first use.

fapaul · 2021-11-03T07:38:06Z

@flinkbot run azure

…ing in LocalStreamEnvironment Before this change, Flink's configuration where not passed to the filesytems if used in LocalStreamEnvironment. This may lead to i.e. missing credentials. This change initializes the filesytems through the pluginloader which forwards the configuration as it is done when running a real cluster.

AHeise

Change looks good. I left one more point to discuss before approval.

AHeise · 2021-11-24T08:15:43Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

 /** MiniCluster to execute Flink jobs locally. */
 public class MiniCluster implements AutoCloseableAsync {

+    public static final ConfigOption<String> PLUGIN_DIRECTORY =


Do we need a config option here? I was thinking that it would be better in this case to add a simple String field to MiniClusterConfiguration iff the option is only ever used in the MiniCluster. Then we can make it easier to use in MiniClusterConfiguration#Builder.

I think this depends a bit if a user only wants to execute a jar locally. In this case one usually does not interact with the minicluster it is only started in the background by the LocalExecutionEnvironment.

Perhaps one option can be to offer both possibilities but this may also confuse users. WDYT?

If you want to support LocalExecutionEnvironment then, we should move the options to somewhere more visible. Then the question arises why we are not supporting all envs ;).

Or we just say that for MiniCluster that we just use MiniClusterConfiguration and for everything else we use the ENV_FLINK_PLUGINS_DIR?

patricklucas · 2023-07-18T10:16:26Z

Would appreciate this change making it in. It's pretty involved to be able to safely use e.g. flink-gs-fs-hadoop locally.

github-actions · 2025-01-15T00:16:15Z

This PR is being marked as stale since it has not had any activity in the last 180 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the [community](https://flink.apache.org/what-is-flink/community/).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 90 days, it will be automatically closed.

github-actions · 2025-03-30T06:25:28Z

This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.

rmetzger added the component=Runtime/Task label Oct 27, 2021

fapaul force-pushed the FLINK-11470 branch from bbad41f to 82f7380 Compare October 28, 2021 11:42

fapaul commented Oct 28, 2021

View reviewed changes

flink-clients/src/test/java/org/apache/flink/client/ExecutionEnvironmentTest.java Outdated Show resolved Hide resolved

AHeise reviewed Nov 1, 2021

View reviewed changes

AHeise requested a review from zentol November 1, 2021 07:59

fapaul force-pushed the FLINK-11470 branch from 82f7380 to 501c93d Compare November 1, 2021 13:23

zentol reviewed Nov 2, 2021

View reviewed changes

flink-clients/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory Outdated Show resolved Hide resolved

zentol requested changes Nov 2, 2021

View reviewed changes

fapaul force-pushed the FLINK-11470 branch 2 times, most recently from 21df3dc to d1682e7 Compare November 2, 2021 16:08

fapaul force-pushed the FLINK-11470 branch from d1682e7 to f8a65a0 Compare November 9, 2021 15:01

AHeise reviewed Nov 24, 2021

View reviewed changes

1996fanrui added the stale label Jan 17, 2025

github-actions bot added the closed-stale label Mar 30, 2025

github-actions bot closed this Mar 30, 2025

[FLINK-11470][runtime] Pass configurations to filesystems when executing in LocalStreamEnvironment #17580

[FLINK-11470][runtime] Pass configurations to filesystems when executing in LocalStreamEnvironment #17580

Uh oh!

Conversation

fapaul commented Oct 27, 2021

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

flinkbot commented Oct 27, 2021

Automated Checks

Review Progress

Uh oh!

flinkbot commented Oct 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

AHeise left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zentol left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fapaul commented Nov 3, 2021

Uh oh!

flinkbot commented Oct 27, 2021 •

edited

Loading