feat (Synchronizer): forward realm-context header on all Polaris requests

polaris-synchronizer/api/src/main/java/org/apache/polaris/tools/sync/polaris/auth/AuthenticationSessionWrapper.java

@@ -43,9 +43,17 @@ public class AuthenticationSessionWrapper implements Closeable {
     private final ScheduledExecutorService executor;
 
     public AuthenticationSessionWrapper(Map<String, String> properties) {
-        this.restClient = HTTPClient.builder(Map.of())
-                .uri(properties.get(OAuth2Properties.OAUTH2_SERVER_URI))
-                .build();
+        HTTPClient.Builder clientBuilder = HTTPClient.builder(Map.of())
+                .uri(properties.get(OAuth2Properties.OAUTH2_SERVER_URI));
+
+        // some Polaris deployments require the realm-context header on the token endpoint
+        // itself, not just on subsequent management/catalog API calls
+        String realm = properties.get("realm");
+        if (realm != null) {
+            clientBuilder.withHeader(properties.getOrDefault("realm-header-name", "Polaris-Realm"), realm);
+        }
+
+        this.restClient = clientBuilder.build();
         this.authSession = this.newAuthSession(this.restClient, properties);
         this.executor = ThreadPools.newScheduledPool(UUID.randomUUID() + "-token-refresh", 1);
     }

polaris-synchronizer/api/src/main/java/org/apache/polaris/tools/sync/polaris/catalog/PolarisCatalog.java

@@ -117,6 +117,13 @@ public Table loadTable(TableIdentifier ident, String etag) {
 
     this.authenticationSession.getSessionHeaders().forEach(requestBuilder::header);
 
+    // this manual HTTP path bypasses RESTSessionCatalog's "header.<name>" property convention,
+    // so the realm-context header must be attached explicitly here as well
+    String realm = this.properties.get("realm");
+    if (realm != null) {
+      requestBuilder.header(this.properties.getOrDefault("realm-header-name", "Polaris-Realm"), realm);
+    }
+
     // specify last known etag in if-none-match header
     if (etag != null) {
       requestBuilder.header(HttpHeaders.IF_NONE_MATCH, etag);

polaris-synchronizer/api/src/main/java/org/apache/polaris/tools/sync/polaris/service/impl/PolarisApiService.java

@@ -83,8 +83,15 @@ public void initialize(Map<String, String> properties) throws Exception {
 
         this.authenticationSession = new AuthenticationSessionWrapper(properties);
 
-        client.setRequestInterceptor(requestBuilder
-                -> authenticationSession.getSessionHeaders().forEach(requestBuilder::header));
+        String realm = properties.get("realm");
+        String realmHeaderName = properties.getOrDefault("realm-header-name", "Polaris-Realm");
+
+        client.setRequestInterceptor(requestBuilder -> {
+            authenticationSession.getSessionHeaders().forEach(requestBuilder::header);
+            if (realm != null) {
+                requestBuilder.header(realmHeaderName, realm);
+            }
+        });
 
         this.baseUrl = baseUrl;
         this.api = new PolarisManagementDefaultApi(client);

polaris-synchronizer/api/src/main/java/org/apache/polaris/tools/sync/polaris/service/impl/PolarisIcebergCatalogService.java

@@ -61,6 +61,14 @@ public PolarisIcebergCatalogService(
                 "credential", String.format("%s:%s", clientId, clientSecret));
         catalogProperties.putIfAbsent("scope", "PRINCIPAL_ROLE:ALL");
 
+        // forward the realm-context header on every Iceberg REST catalog call, leveraging
+        // RESTSessionCatalog's built-in "header.<name>" -> HTTP header property convention
+        String realm = properties.get("realm");
+        if (realm != null) {
+            String realmHeaderName = properties.getOrDefault("realm-header-name", "Polaris-Realm");
+            catalogProperties.put("header." + realmHeaderName, realm);
+        }
+
         this.catalog = (PolarisCatalog) CatalogUtil.loadCatalog(
                 PolarisCatalog.class.getName(),
                 "SOURCE_CATALOG_REST_" + catalogName,

polaris-synchronizer/cli/src/main/java/org/apache/polaris/tools/sync/polaris/CLIUtil.java

@@ -29,7 +29,9 @@ public class CLIUtil {
             "\n\t- token: the bearer token to authenticate against the Polaris instance with." +
             "\n\t- oauth2-server-uri: the uri of the OAuth2 server to authenticate to. (eg. http://localhost:8181/api/catalog/v1/oauth/tokens)" +
             "\n\t- credential: the client credentials to use to authenticate against the Polaris instance (eg. <client_id>:client_secret>)" +
-            "\n\t- scope: the scope to authenticate with for the service_admin (eg. PRINCIPAL_ROLE:ALL)";
+            "\n\t- scope: the scope to authenticate with for the service_admin (eg. PRINCIPAL_ROLE:ALL)" +
+            "\n\t- realm: the realm to send in the realm-context header on every request to the Polaris instance (eg. POLARIS). If unset, no realm header is sent." +
+            "\n\t- realm-header-name: (default: Polaris-Realm) the name of the header used to convey the realm; must match the Polaris instance's configured polaris.realm-context.header-name";
 
     public static final String OMNIPOTENT_PRINCIPAL_PROPERTIES_DESCRIPTION =
             "\nOmnipotent Principal Properties:" +