This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 189b6ca8313 [FLINK-40592][metrics] Fix PushGateway basic
authentication without JAXB (#29151)
189b6ca8313 is described below
commit 189b6ca8313ada196596cb53315866e7a531d441
Author: weicheng-07 <[email protected]>
AuthorDate: Mon Sep 14 02:44:17 2026 +0800
[FLINK-40592][metrics] Fix PushGateway basic authentication without JAXB
(#29151)
Encode the Basic Authorization header with JDK Base64 and UTF-8 while
preserving the default connection factory and credential configuration.
Cover PUT and DELETE requests with regression tests that isolate JAXB.
Generated-by: OpenAI Codex 0.153.4
Co-authored-by: jiweicheng <[email protected]>
---
.../prometheus/PrometheusPushGatewayReporter.java | 30 ++++-
...theusPushGatewayReporterAuthenticationTest.java | 146 +++++++++++++++++++++
2 files changed, 174 insertions(+), 2 deletions(-)
diff --git
a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
index 2ee6d6df431..9de72da28e5 100644
---
a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
+++
b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporter.java
@@ -25,13 +25,17 @@ import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.metrics.reporter.Scheduled;
import org.apache.flink.util.Preconditions;
-import io.prometheus.client.exporter.BasicAuthHttpConnectionFactory;
+import io.prometheus.client.exporter.DefaultHttpConnectionFactory;
+import io.prometheus.client.exporter.HttpConnectionFactory;
import io.prometheus.client.exporter.PushGateway;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.net.HttpURLConnection;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
import java.util.Map;
/**
@@ -58,7 +62,7 @@ public class PrometheusPushGatewayReporter extends
AbstractPrometheusReporter im
this.pushGateway = new PushGateway(hostUrl);
if (username != null && password != null) {
this.pushGateway.setConnectionFactory(
- new BasicAuthHttpConnectionFactory(username, password));
+ new JdkBasicAuthHttpConnectionFactory(username, password));
this.basicAuthEnabled = true;
} else {
this.basicAuthEnabled = false;
@@ -96,4 +100,26 @@ public class PrometheusPushGatewayReporter extends
AbstractPrometheusReporter im
}
super.close();
}
+
+ private static final class JdkBasicAuthHttpConnectionFactory implements
HttpConnectionFactory {
+ private final HttpConnectionFactory connectionFactory = new
DefaultHttpConnectionFactory();
+ private final String authorizationHeader;
+
+ private JdkBasicAuthHttpConnectionFactory(String username, String
password) {
+ // Use the JDK encoder because simpleclient's implementation
requires JAXB.
+ authorizationHeader =
+ "Basic "
+ + Base64.getEncoder()
+ .encodeToString(
+ (username + ":" + password)
+
.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public HttpURLConnection create(String url) throws IOException {
+ final HttpURLConnection connection = connectionFactory.create(url);
+ connection.setRequestProperty("Authorization",
authorizationHeader);
+ return connection;
+ }
+ }
}
diff --git
a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterAuthenticationTest.java
b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterAuthenticationTest.java
new file mode 100644
index 00000000000..633f1bfc922
--- /dev/null
+++
b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterAuthenticationTest.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.metrics.prometheus;
+
+import org.apache.flink.metrics.MetricConfig;
+import org.apache.flink.metrics.reporter.MetricReporter;
+import org.apache.flink.metrics.reporter.MetricReporterFactory;
+import org.apache.flink.metrics.reporter.Scheduled;
+
+import com.sun.net.httpserver.HttpServer;
+import io.prometheus.client.Collector;
+import io.prometheus.client.exporter.PushGateway;
+import io.prometheus.client.exporter.common.TextFormat;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.DELETE_ON_SHUTDOWN;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.HOST_URL;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.JOB_NAME;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.PASSWORD;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.RANDOM_JOB_NAME_SUFFIX;
+import static
org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.USERNAME;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests authentication on actual PushGateway requests. */
+class PrometheusPushGatewayReporterAuthenticationTest {
+
+ @ParameterizedTest
+ @CsvSource({
+ "user, password, Basic dXNlcjpwYXNzd29yZA==",
+ "user, päss:word, Basic dXNlcjpww6Rzczp3b3Jk",
+ "'', '', Basic Og==",
+ ", ,",
+ "user, ,",
+ ", password,"
+ })
+ void testAuthenticationOnPushAndDeleteWithoutJaxb(
+ String username, String password, String expectedAuthorization)
throws Exception {
+ final List<String> methods = Collections.synchronizedList(new
ArrayList<>());
+ final List<String> authorizations = Collections.synchronizedList(new
ArrayList<>());
+ final HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext(
+ "/metrics/job/test-job",
+ exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ methods.add(exchange.getRequestMethod());
+
authorizations.add(exchange.getRequestHeaders().getFirst("Authorization"));
+ exchange.sendResponseHeaders(202, -1);
+ } finally {
+ exchange.close();
+ }
+ });
+ server.start();
+ try (URLClassLoader classLoader = createClassLoaderWithoutJaxb()) {
+ final MetricConfig config = new MetricConfig();
+ config.setProperty(HOST_URL.key(), "http://127.0.0.1:"; +
server.getAddress().getPort());
+ config.setProperty(JOB_NAME.key(), "test-job");
+ config.setProperty(RANDOM_JOB_NAME_SUFFIX.key(), "false");
+ config.setProperty(DELETE_ON_SHUTDOWN.key(), "true");
+ if (username != null) {
+ config.setProperty(USERNAME.key(), username);
+ }
+ if (password != null) {
+ config.setProperty(PASSWORD.key(), password);
+ }
+ final MetricReporterFactory factory =
+ (MetricReporterFactory)
+ classLoader
+
.loadClass(PrometheusPushGatewayReporterFactory.class.getName())
+ .getDeclaredConstructor()
+ .newInstance();
+ final MetricReporter reporter =
factory.createMetricReporter(config);
+ try {
+ assertThatThrownBy(
+ () ->
+ reporter.getClass()
+ .getClassLoader()
+
.loadClass("javax.xml.bind.DatatypeConverter"))
+ .isInstanceOf(ClassNotFoundException.class);
+ ((Scheduled) reporter).report();
+ } finally {
+ reporter.close();
+ }
+ assertThat(methods).containsExactly("PUT", "DELETE");
+ assertThat(authorizations)
+ .containsExactly(expectedAuthorization,
expectedAuthorization);
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private static URLClassLoader createClassLoaderWithoutJaxb() {
+ final URL[] urls =
+ Stream.of(
+ PrometheusPushGatewayReporter.class,
+ Collector.class,
+ PushGateway.class,
+ TextFormat.class)
+ .map(type ->
type.getProtectionDomain().getCodeSource().getLocation())
+ .distinct()
+ .toArray(URL[]::new);
+ // Reload the reporter and its Prometheus dependencies so JAXB on the
test classpath
+ // cannot hide the missing runtime dependency.
+ final ClassLoader parent =
+ new ClassLoader(
+
PrometheusPushGatewayReporterAuthenticationTest.class.getClassLoader()) {
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve)
+ throws ClassNotFoundException {
+ if (name.startsWith("javax.xml.bind.")
+ ||
name.startsWith("org.apache.flink.metrics.prometheus.")
+ || name.startsWith("io.prometheus.")) {
+ throw new ClassNotFoundException(name);
+ }
+ return super.loadClass(name, resolve);
+ }
+ };
+ return new URLClassLoader(urls, parent);
+ }
+}
