[FLINK-37985][conf] Refactor public config options from StreamConfig to JobConfiguration

[1996fanrui]

What is the purpose of the change

Some public config options are stored in StreamConfig, StreamConfig stores some operator related information, such as class serialization result, vertex id, etc. But storing public config options are confusing.

It's better to refactor them from StreamConfig to JobConfiguration. And all callers could read these config options from JobConfigruration instead of StreamConfig.

Brief change log

• [FLINK-37985][hotfix] Cleanup unused fields and methods in StreamConfig
• [FLINK-37985][conf] Refactor public config options from StreamConfig to JobConfiguration
• [FLINK-37985][conf] Extract some static methods in CheckpointingOptions to simplify read config options
• [FLINK-37985][conf] Refactor CheckpointingMode from StreamConfig to JobConfiguration
• [FLINK-37985][conf] Refactor CheckpointingEnabled from StreamConfig to JobConfiguration
• [FLINK-37985][test] Introduce the arch rules to prevent some checkpoint config options are got directly from Configuration.get() or Configuration.getOptional()

Verifying this change

• Improved tons of existing tests
• Introducing a new test class: CheckpointingOptionsTest
• Introducing CheckpointingConfigurationAccessRules to prevent some checkpoint config options are got directly from Configuration.get() or Configuration.getOptional()

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

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

Documentation

• Does this pull request introduce a new feature? no

[flinkbot]

CI report:

• 1650bfb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[1996fanrui]

FYI: After this PR, there are still 2 checkpoint related config options are stored in StreamConfig: CHECKPOINTING_ENABLED and CHECKPOINT_MODE .

The reason for keeping CHECKPOINTING_ENABLED is : Flink doesn't have public config options to indicate CHECKPOINTING_ENABLED, the checkpoint is enable when checkpoint interval > 0 [1].

The reason for keeping CHECKPOINT_MODE is : the checkpoint mode will be AT_LEAST_ONCE when checkpoint is disabled [2] , and it does't update the value in Configuration.

Even so, I still agree to move them outof StreamConfig in the future, we could introduce some static methods in CheckpointingOptions to address these cases, such as:

    public static boolean isCheckpointingEnabled(Configuration config) {
        return config
                .getOptional(CheckpointingOptions.CHECKPOINTING_INTERVAL)
                .map(Duration::toMillis)
                .orElse(-1L)
                > 0;
    }

    public static CheckpointingMode getCheckpointingMode(Configuration config) {
        if (isCheckpointingEnabled(config)) {
            return config.get(CHECKPOINTING_CONSISTENCY_MODE);
        } else {
            // the "at-least-once" input handler is slightly cheaper (in the absence of
            // checkpoints), so we use that one if checkpointing is not enabled
            return CheckpointingMode.AT_LEAST_ONCE;
        }
    }

It works similarly to some scheduler config options[3].

[1]

flink/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java

Line 100 in df5bdc6

public boolean isCheckpointingEnabled() {

[2]

flink/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java

Line 254 in df5bdc6

public static CheckpointingMode getCheckpointingMode(CheckpointConfig checkpointConfig) {

[3]

flink/flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java

Line 177 in df5bdc6

public static JobManagerOptions.SchedulerType getSchedulerType(Configuration configuration) {

[rkhachatryan]

Thanks for the PR @1996fanrui, the changes LGTM.

Could you explain why don't you replace the two remaining options though?
I think it would be more consistent to have the same way of handling all checkpointing options.

Even so, I still agree to move them outof StreamConfig in the future, we could introduce some static methods in CheckpointingOptions to address these cases

Or would it be a much bigger change?

[1996fanrui]

Thanks @rkhachatryan for the quick review!

Could you explain why don't you replace the two remaining options though? Or would it be a much bigger change?

It has 2 reasons: bigger change and need to introduce some static methods to read these config options(And ensure no other callers and no new callers to read them from Configuration directly)

I think it would be more consistent to have the same way of handling all checkpointing options.

Yes, I agree with you. I have updated this PR, and solved all of above problems.

Besides of that, I introduced CheckpointingConfigurationAccessRules to prevent some checkpoint config options are got directly from Configuration.get() or Configuration.getOptional(). CI will be failed if these config options are get or getOptional from Configuration directly in the future.

[rkhachatryan]

Thanks for updating the PR @1996fanrui

I think that's a good idea to add arch tests.

I have a couple of minor remarks, otherwise LGTM.

[rkhachatryan]

Should this be removed now?:)

[1996fanrui]

Good catch, and removed!

[rkhachatryan]

It seems possible to use code constants here (and for class names), right?

So that renaming doesn't make this test a no_op

[1996fanrui]

Good suggestion!

Class can be easily converted into a String name, for example: CheckpointingOptions.class.getName();. But Java variable and method names are difficult to convert directly into String names. I used reflection to load variables and methods, and an exception will be thrown if the loading fails. (This arch test will be failed if these variables or methods are removed in the future.)

Following is a demo:

    private static Set<String> getProhibitedFieldNames() {
        try {
            return new HashSet<>(
                    Arrays.asList(
                            CheckpointingOptions.class.getField("ENABLE_UNALIGNED").getName(),
                            CheckpointingOptions.class
                                    .getField("CHECKPOINTING_CONSISTENCY_MODE")
                                    .getName(),
                            CheckpointingOptions.class
                                    .getField("ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS")
                                    .getName()));
        } catch (NoSuchFieldException e) {
            // This makes the test class fail to load if a field is ever renamed.
            throw new ExceptionInInitializerError(e);
        }
    }

[rkhachatryan]

LGTM

[1996fanrui]

@flinkbot run azure

[1996fanrui]

Thanks for the review, merging