public void configure(ReadableConfig configuration, ClassLoader classLoader) {
        configuration
                .getOptional(PipelineOptions.AUTO_TYPE_REGISTRATION)
                .ifPresent(this::setAutoTypeRegistration);
        configuration
                .getOptional(PipelineOptions.AUTO_GENERATE_UIDS)
                .ifPresent(this::setAutoGeneratedUids);
        configuration
                .getOptional(PipelineOptions.AUTO_WATERMARK_INTERVAL)
                .ifPresent(this::setAutoWatermarkInterval);
        configuration
                .getOptional(PipelineOptions.CLOSURE_CLEANER_LEVEL)
                .ifPresent(this::setClosureCleanerLevel);
        configuration
                .getOptional(PipelineOptions.GLOBAL_JOB_PARAMETERS)
                .ifPresent(this::setGlobalJobParameters);

        configuration
                .getOptional(MetricOptions.LATENCY_INTERVAL)
                .ifPresent(interval -> setLatencyTrackingInterval(interval.toMillis()));

        configuration
                .getOptional(StateChangelogOptions.PERIODIC_MATERIALIZATION_ENABLED)
                .ifPresent(this::enablePeriodicMaterialize);
        configuration
                .getOptional(StateChangelogOptions.PERIODIC_MATERIALIZATION_INTERVAL)
                .ifPresent(this::setPeriodicMaterializeIntervalMillis);
        configuration
                .getOptional(StateChangelogOptions.MATERIALIZATION_MAX_FAILURES_ALLOWED)
                .ifPresent(this::setMaterializationMaxAllowedFailures);

        configuration
                .getOptional(PipelineOptions.MAX_PARALLELISM)
                .ifPresent(this::setMaxParallelism);
        configuration.getOptional(CoreOptions.DEFAULT_PARALLELISM).ifPresent(this::setParallelism);
        configuration.getOptional(PipelineOptions.OBJECT_REUSE).ifPresent(this::setObjectReuse);
        configuration
                .getOptional(TaskManagerOptions.TASK_CANCELLATION_INTERVAL)
                .ifPresent(interval -> setTaskCancellationInterval(interval.toMillis()));
        configuration
                .getOptional(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT)
                .ifPresent(timeout -> setTaskCancellationTimeout(timeout.toMillis()));
        configuration
                .getOptional(ExecutionOptions.SNAPSHOT_COMPRESSION)
                .ifPresent(this::setUseSnapshotCompression);
        configuration
                .getOptional(RestartStrategyOptions.RESTART_STRATEGY)
                .ifPresent(
                        s -> {
                            this.setRestartStrategy(configuration);
                            // reset RestartStrategies for backward compatibility
                            this.setRestartStrategy(
                                    new RestartStrategies.FallbackRestartStrategyConfiguration());
                        });

        configuration
                .getOptional(JobManagerOptions.SCHEDULER)
                .ifPresent(t -> this.configuration.set(JobManagerOptions.SCHEDULER, t));

        serializerConfig.configure(configuration, classLoader);
    }
@Test
    void testReadingDefaultConfig() {
        ExecutionConfig executionConfig = new ExecutionConfig();
        Configuration configuration = new Configuration();

        // mutate config according to configuration
        executionConfig.configure(configuration, ExecutionConfigTest.class.getClassLoader());

        assertThat(executionConfig).isEqualTo(new ExecutionConfig());
    }