[rest] use baseHeaders instead of emptyMap in RESTApi GetConfig (#6410)

Author: plusplusjiajia

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -92,7 +92,6 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -192,8 +191,7 @@ public RESTApi(Options options, boolean configRequired) {
                                             ResourcePaths.config(),
                                             queryParams,
                                             ConfigResponse.class,
-                                            new RESTAuthFunction(
-                                                    Collections.emptyMap(), authProvider))
+                                            new RESTAuthFunction(baseHeaders, authProvider))
                                     .merge(options.toMap()));
             baseHeaders.putAll(extractPrefixMap(options, HEADER_PREFIX));
         }

paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java

@@ -170,6 +170,48 @@ void testHeaderOptions() throws Exception {
         assertEquals(headers2.get("User-Agent"), "test");
     }
 
+    @Test
+    void testBaseHeadersInRequests() throws Exception {
+        // Set custom headers in options
+        String customHeaderName = "custom-header";
+        String customHeaderValue = "custom-value";
+        options.set(HEADER_PREFIX + customHeaderName, customHeaderValue);
+
+        // Clear any previous headers
+        restCatalogServer.clearReceivedHeaders();
+        assertEquals(0, restCatalogServer.getReceivedHeaders().size());
+
+        // Initialize catalog with custom headers
+        RESTCatalog restCatalog = initCatalog(false);
+        // init catalog will trigger REST GetConfig request
+        checkHeader(customHeaderName, customHeaderValue);
+
+        // Clear any previous headers
+        restCatalogServer.clearReceivedHeaders();
+        assertEquals(0, restCatalogServer.getReceivedHeaders().size());
+
+        // Perform an operation that will trigger REST request
+        restCatalog.listDatabases();
+        checkHeader(customHeaderName, customHeaderValue);
+    }
+
+    private void checkHeader(String headerName, String headerValue) {
+        // Verify that the header were included in the requests
+        List<Map<String, String>> receivedHeaders = restCatalogServer.getReceivedHeaders();
+        assert receivedHeaders.size() > 0 : "No requests were recorded";
+
+        // Check that request contains our custom headers
+        boolean foundCustomHeader = false;
+
+        for (Map<String, String> headers : receivedHeaders) {
+            if (headerValue.equals(headers.get(headerName))) {
+                foundCustomHeader = true;
+            }
+        }
+
+        assert foundCustomHeader : "Header was not found in any request";
+    }
+
     private void testDlfAuth(RESTCatalog restCatalog) throws Exception {
         String databaseName = "db1";
         restCatalog.createDatabase(databaseName, true);

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java

@@ -180,6 +180,8 @@ public class RESTCatalogServer {
 
     private final ResourcePaths resourcePaths;
 
+    private final List<Map<String, String>> receivedHeaders = new ArrayList<>();
+
     public RESTCatalogServer(
             String dataPath, AuthProvider authProvider, ConfigResponse config, String warehouse) {
         this.warehouse = warehouse;
@@ -275,6 +277,7 @@ public MockResponse dispatch(RecordedRequest request) {
                 RESTResponse response;
                 try {
                     Map<String, String> headers = getHeader(request);
+                    receivedHeaders.add(new HashMap<>(headers));
                     String[] paths = request.getPath().split("\\?");
                     String resourcePath = paths[0];
                     Map<String, String> parameters =
@@ -2375,4 +2378,12 @@ private MockResponse handleInvalidMaxResults(Map<String, String> parameters) {
                         400),
                 400);
     }
+
+    public List<Map<String, String>> getReceivedHeaders() {
+        return receivedHeaders;
+    }
+
+    public void clearReceivedHeaders() {
+        receivedHeaders.clear();
+    }
 }