[FLINK-11193][State Backends]Use user passed configuration overriding default configuration loading from file #8479

Aitozi · 2019-05-19T12:37:33Z

What is the purpose of the change

As described in the jira, User's customize configuration which is configured by backend.configure() method will be override by the configuration loading from flink-conf.yaml. I think the config in the code should has a higher priority than the default file configuration.

Brief change log

Merge configuration before create the new state backend.

Verifying this change

Adding a test case to verify.

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (100MB)
Extended integration test for recovery after master (JobManager) failure
Added test that validates that TaskInfo is transferred only once across recoveries
Manually verified the change by running a 4 node cluser with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

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, Yarn/Mesos, 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 · 2019-05-19T12:38:39Z

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 2e316e2 (Fri Feb 28 21:49:54 UTC 2020)

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

Aitozi · 2019-05-19T12:39:18Z

Hi, @StefanRRichter can you help take a look on this PR when you have time, thx.

Aitozi · 2019-05-20T02:21:35Z

The travis-ci failure seems unrelated

klion26 · 2019-05-22T04:18:49Z

...kend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java

 	 * @param classLoader The class loader.
 	 */
 	private RocksDBStateBackend(RocksDBStateBackend original, Configuration config, ClassLoader classLoader) {
+		this.configuration = config;


When do we use RocksDBStateBackend, seems we always use the original statebackend's conf first?

Thanks @klion26 for your review. I'm not sure whether i quite get you meaning. But IMO this construct is only for creating a re-configured copy of the original state backend. So the configure passed in will be used first ? And also should keep this configure to a member field to reuse it to config the statebackend loaded from StateBackendLoader#fromApplicationOrConfigOrDefault.

But the code below(start from line 302) seems to use the original backend's conf first, so I'm not sure you change now is on the right directory(maybe the change you've made can't solve the problem on the issue)?

Hi, @klion26 I check the code again, only the priorityQueueStateType is configured from configuration, and the type from configuration will be used first see

final String priorityQueueTypeString = config.getString(TIMER_SERVICE_FACTORY); this.priorityQueueStateType = priorityQueueTypeString.length() > 0 ? PriorityQueueStateType.valueOf(priorityQueueTypeString.toUpperCase()) : original.priorityQueueStateType;

And other parameters are all passed by the constructor, so it can work for the issue i think. but it looks strange indeed(some follow the original backend, some follow the config first), I will think how to make it better.

How about the other configs? Do we need to do the same thing here for the other configs?

other config are all following this manner: check the original config whether set or undefined, if it is not defined then it will look into the config , at last fallback to the default config which is consistent with the doc of the config()

Creates a copy of this state backend that uses the values defined in the configuration for fields where that were not yet specified in this state backend.

But for the priorityQueueStateType there is no undefined type, it only has two: rocksdb and heap, if we have to follow the manner I think we just have to add an undefined value for priorityQueueStateType, but i am not sure do we need to do this.

What i am do now is just overriding the config loading from file with the original statebackend config to solve this problem.

Do you think we need to check the config first, then fallback the original config, and at last fallback to the default config?

I think we don't have to, I prefer to keep the former style, but only use the config in user code with higher priority than the default configuration loading from the file.

Aitozi · 2019-06-02T13:33:32Z

Hi @klion26 I refactor the test case to verify this fix work explicitly. @StefanRRichter could you help review this bug fix, too.

flinkbot · 2019-08-30T08:56:31Z

CI report:

2e316e2 Travis: SUCCESS Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

azagrebin

Thanks for looking into this problem @Aitozi
I think you are right that the user custom configuration should have a precedence.
However, it looks like other parameters already have a certain configuration pattern which was probably accidentally overlooked in the original effort.
E.g. localRocksDbDirectories is firstly checked whether it is already set or not by user. If it is set then it is not overwritten otherwise it is set from the config which will be default value in case of cluster config.
It looks like it would be simpler just to stick to the same pattern for priorityQueueStateType instead of changing the existing approach.
wdyt?

Aitozi · 2019-09-02T09:49:33Z

Thanks for your suggestion @azagrebin , I will look into this .

Aitozi · 2019-09-04T15:00:13Z

Hi @azagrebin , I have done the fix following your comments, please help review when you are free.

Aitozi · 2019-09-04T23:27:31Z

Do you know why this push do not trigger travis ci ? How can i trigger it manually @azagrebin ?

azagrebin · 2019-09-05T12:16:23Z

@Aitozi it triggered the CI, the setup has been recently changed, you can see CI results in the special build comment in this PR.

Aitozi · 2019-09-05T12:31:15Z

@azagrebin I check the CI log just now, I think it failed due to

PackagesNotFoundError: The following packages are not available from current channels:

python=3.3

seems unrelated to this PR ?

azagrebin

Thanks for working on this @Aitozi
I left some smaller comments

azagrebin · 2019-09-05T12:53:30Z

...kend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java

-			PriorityQueueStateType.valueOf(priorityQueueTypeString.toUpperCase()) : original.priorityQueueStateType;
+		if (original.priorityQueueStateType == PriorityQueueStateType.UNDEFINED) {
+			String priorityQueueTypeString = config.getString(TIMER_SERVICE_FACTORY);
+			this.priorityQueueStateType = PriorityQueueStateType.valueOf(priorityQueueTypeString.toUpperCase());


short-cut for getting Enum values is: config.getEnum(...)

azagrebin · 2019-09-05T13:00:53Z

...kend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java

-			PriorityQueueStateType.valueOf(priorityQueueTypeString.toUpperCase()) : original.priorityQueueStateType;
+		if (original.priorityQueueStateType == PriorityQueueStateType.UNDEFINED) {
+			String priorityQueueTypeString = config.getString(TIMER_SERVICE_FACTORY);
+			this.priorityQueueStateType = PriorityQueueStateType.valueOf(priorityQueueTypeString.toUpperCase());


I think introducing UNDEFINED makes sense in general but I am wondering if making it @nullable and leaving initialized by null is actually simpler in this case. If we use UNDEFINED, I think we also have to check here explicitly and then test it that if user configures UNDEFINED then she gets an error at once.

agree to use null directly.

azagrebin · 2019-09-05T13:03:00Z

...kend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java

 		enableTtlCompactionFilter = TernaryBoolean.TRUE;
 	}

+	private PriorityQueueStateType resolvePriorityQueueStateType(PriorityQueueStateType origin) {


I am wondering if it is more user-friendly if we add setter/getter for this like we have for some other things.
we can also use this getter with the resolution logic in createKeyedStateBackend.

azagrebin · 2019-09-05T13:03:48Z