Repository: apache/flink
File: flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterFactory.java (Java)

Code before:
            throw new IllegalArgumentException("hostUrl must not be null or empty");
        }

        String jobName = configuredJobName;
        if (randomSuffix) {
            jobName = configuredJobName + new AbstractID();
        }

        String username = metricConfig.getString(USERNAME.key(), USERNAME.defaultValue());
        String password = metricConfig.getString(PASSWORD.key(), PASSWORD.defaultValue());

        if ((username != null && password == null) || (username == null && password != null)) {
            LOG.warn(
                    "Both username and password must be configured to enable HTTP Basic Authentication. "
                            + "Currently only {} is configured, Basic Auth will be disabled.",
                    username != null ? "username" : "password");
        }

        LOG.info(
                "Configured PrometheusPushGatewayReporter with {hostUrl:{}, jobName:{}, randomJobNameSuffix:{}, deleteOnShutdown:{}, groupingKey:{}, basicAuth:{}}",
                hostUrl,
                jobName,
                randomSuffix,
                deleteOnShutdown,
                groupingKey,
                username != null && password != null);

        try {
            return new PrometheusPushGatewayReporter(
                    new URL(hostUrl), jobName, groupingKey, deleteOnShutdown, username, password);
        } catch (MalformedURLException e) {
            throw new RuntimeException(e);
        }
    }

    @VisibleForTesting
    static Map<String, String> parseGroupingKey(final String groupingKeyConfig) {
        if (!groupingKeyConfig.isEmpty()) {
            Map<String, String> groupingKey = new HashMap<>();
            String[] kvs = groupingKeyConfig.split(";");
            for (String kv : kvs) {
                int idx = kv.indexOf("=");
                if (idx < 0) {
                    LOG.warn("Invalid prometheusPushGateway groupingKey:{}, will be ignored", kv);
                    continue;
                }

                String labelKey = kv.substring(0, idx);
                String labelValue = kv.substring(idx + 1);
                if (StringUtils.isNullOrWhitespaceOnly(labelKey)
                        || StringUtils.isNullOrWhitespaceOnly(labelValue)) {

Review comment (security):
Minor nit: this can probably align with the recently added `basicAuthEnabled` flag to avoid the redundant logic.

Code after:
            throw new IllegalArgumentException("hostUrl must not be null or empty");
        }

        String jobName = configuredJobName;
        if (randomSuffix) {
            jobName = configuredJobName + new AbstractID();
        }

        String username = metricConfig.getString(USERNAME.key(), USERNAME.defaultValue());
        String password = metricConfig.getString(PASSWORD.key(), PASSWORD.defaultValue());

        if ((username != null && password == null) || (username == null && password != null)) {
            LOG.warn(
                    "Both username and password must be configured to enable HTTP Basic Authentication. "
                            + "Currently only {} is configured, Basic Auth will be disabled.",
                    username != null ? "username" : "password");
        }

        try {
            PrometheusPushGatewayReporter reporter =
                    new PrometheusPushGatewayReporter(
                            new URL(hostUrl),
                            jobName,
                            groupingKey,
                            deleteOnShutdown,
                            username,
                            password);

            LOG.info(
                    "Configured PrometheusPushGatewayReporter with {hostUrl:{}, jobName:{}, randomJobNameSuffix:{}, deleteOnShutdown:{}, groupingKey:{}, basicAuth:{}}",
                    hostUrl,
                    jobName,
                    randomSuffix,
                    deleteOnShutdown,
                    groupingKey,
                    reporter.basicAuthEnabled);

            return reporter;
        } catch (MalformedURLException e) {
            throw new RuntimeException(e);
        }
    }

    @VisibleForTesting
    static Map<String, String> parseGroupingKey(final String groupingKeyConfig) {
        if (!groupingKeyConfig.isEmpty()) {
            Map<String, String> groupingKey = new HashMap<>();
            String[] kvs = groupingKeyConfig.split(";");
            for (String kv : kvs) {
                int idx = kv.indexOf("=");
                if (idx < 0) {