[SPARK-51770] Set Content-Type headers for Prometheus v3

### What changes were proposed in this pull request?

This PR sets Content-Type header for Prometheus handler to make it compatible with Prometheus v3

### Why are the changes needed?

Prometheus v3 will fail the scrape if the Content-Type header is not present in response and the fallback scrape protocol is not configured. Therefore we'd configure the headers properly.

Please also refer https://prometheus.io/docs/prometheus/latest/migration/#scrape-protocols

### Does this PR introduce any user-facing change?

No

### How was this patch tested?

Tested via CI and local set up for Prometheus v3, via steps:

* Deploy Prometheus v3 in k8s cluster, with default configuration (no fallback_scrape_protocol configured)
* Deploy Spark Operator in cluster
* Prometheus v3 is able to successfully scrape operator metrics

### Was this patch authored or co-authored using generative AI tooling?

No

Closes apache#171 from jiangzho/monitoring.

Authored-by: Zhou JIANG <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: jiangzho

spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/PrometheusPullModelHandler.java

@@ -24,7 +24,9 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 
 import com.codahale.metrics.MetricRegistry;
@@ -58,7 +60,11 @@ public void stop() {
   public void handle(HttpExchange exchange) throws IOException {
     HttpServletRequest httpServletRequest = null;
     String value = getMetricsSnapshot(httpServletRequest);
-    sendMessage(exchange, HTTP_OK, String.join("\n", filterNonEmptyRecords(value)));
+    sendMessage(
+        exchange,
+        HTTP_OK,
+        String.join("\n", filterNonEmptyRecords(value)),
+        Map.of("Content-Type", Collections.singletonList("text/plain;version=0.0.4")));
   }
 
   protected List<String> filterNonEmptyRecords(String metricsSnapshot) {

spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ProbeUtil.java

@@ -23,8 +23,10 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
+import com.sun.net.httpserver.Headers;
 import com.sun.net.httpserver.HttpExchange;
 import io.javaoperatorsdk.operator.Operator;
 import io.javaoperatorsdk.operator.RuntimeInfo;
@@ -46,8 +48,28 @@ private ProbeUtil() {}
    */
   public static void sendMessage(HttpExchange httpExchange, int code, String message)
       throws IOException {
+    sendMessage(httpExchange, code, message, null);
+  }
+
+  /**
+   * Send an HTTP response message with the given response header HTTP status code, message and
+   * headers.
+   *
+   * @param httpExchange The handler for this HTTP response.
+   * @param code A response header HTTP status code defined in java.net.HttpURLConnection.HTTP_*
+   * @param message A message to send as a body
+   * @param headers Headers to be sent with the response
+   * @throws IOException Failed to send a response.
+   */
+  public static void sendMessage(
+      HttpExchange httpExchange, int code, String message, Map<String, List<String>> headers)
+      throws IOException {
     try (OutputStream outputStream = httpExchange.getResponseBody()) {
       byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
+      if (headers != null && !headers.isEmpty()) {
+        Headers responseHeaders = httpExchange.getResponseHeaders();
+        responseHeaders.putAll(headers);
+      }
       httpExchange.sendResponseHeaders(code, bytes.length);
       outputStream.write(bytes);
       outputStream.flush();