@Override
    public Configuration toConfiguration(CommandLine commandLine) throws FlinkException {

        final Configuration resultingConfiguration = super.toConfiguration(commandLine);
        if (commandLine.hasOption(addressOption.getOpt())) {
            String addressWithPort = commandLine.getOptionValue(addressOption.getOpt());
            InetSocketAddress jobManagerAddress = NetUtils.parseHostPortAddress(addressWithPort);
            setJobManagerAddressInConfig(resultingConfiguration, jobManagerAddress);

            URL url = NetUtils.getCorrectHostnamePort(addressWithPort);
            resultingConfiguration.set(RestOptions.PATH, url.getPath());
            resultingConfiguration.set(SecurityOptions.SSL_REST_ENABLED, isHttpsProtocol(url));
        }
        resultingConfiguration.set(DeploymentOptions.TARGET, RemoteExecutor.NAME);

        DynamicPropertiesUtil.encodeDynamicProperties(commandLine, resultingConfiguration);

        return resultingConfiguration;
    }
@Test
    void testDynamicPropertyMaterialization() throws Exception {
        final String[] args = {
            "-D" + PipelineOptions.AUTO_WATERMARK_INTERVAL.key() + "=42",
            "-D" + PipelineOptions.AUTO_GENERATE_UIDS.key() + "=true"
        };

        final AbstractCustomCommandLine defaultCLI = new DefaultCLI();
        final CommandLine commandLine = defaultCLI.parseCommandLineOptions(args, false);

        Configuration configuration = defaultCLI.toConfiguration(commandLine);

        assertThat(configuration.get(PipelineOptions.AUTO_WATERMARK_INTERVAL))
                .isEqualTo(Duration.ofMillis(42L));
        assertThat(configuration.get(PipelineOptions.AUTO_GENERATE_UIDS)).isTrue();
    }