diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index b6cc699a..baf70129 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -33,4 +33,4 @@ jobs:
             ${{ runner.os }}-maven-
       - name: Build with Maven
         # don't run ConnectionTest for now.
-        run: mvn -B package --file pom.xml -Dtest=TunnelContractsTests
+        run: mvn -B package --file pom.xml -Dtest=TunnelContractsTests,TunnelClusterRecommendationsTests
diff --git a/java/src/main/java/com/microsoft/tunnels/management/ITunnelManagementClient.java b/java/src/main/java/com/microsoft/tunnels/management/ITunnelManagementClient.java
index 70b85e99..d41e7753 100644
--- a/java/src/main/java/com/microsoft/tunnels/management/ITunnelManagementClient.java
+++ b/java/src/main/java/com/microsoft/tunnels/management/ITunnelManagementClient.java
@@ -4,6 +4,7 @@
 package com.microsoft.tunnels.management;
 
 import com.microsoft.tunnels.contracts.ClusterDetails;
+import com.microsoft.tunnels.contracts.ClusterRecommendationResponse;
 import com.microsoft.tunnels.contracts.NamedRateStatus;
 import com.microsoft.tunnels.contracts.Tunnel;
 import com.microsoft.tunnels.contracts.TunnelConnectionMode;
@@ -251,4 +252,21 @@ public CompletableFuture<Boolean> checkNameAvailabilityAsync(
    * @return Array of {@link NamedRateStatus}.
    */
   public CompletableFuture<Collection<NamedRateStatus>> listUserLimitsAsync();
+
+  /**
+   * Requests cluster recommendations for placing a new tunnel, ranked by preference.
+   *
+   * <p>The request is authenticated with the token from the user token callback when one is
+   * configured. If the service rejects that token with a 401 or 403, the request is retried
+   * once anonymously, because the service rejects a bad token before evaluating the request
+   * rather than falling back to treating the caller as anonymous on its own.</p>
+   *
+   * @param preferredClusterId Preferred cluster ID, or null for no preference.
+   * @param requiredGeo        Optional Azure geography filter; only clusters in this geo are
+   *                           eligible for recommendation, or null for no filter.
+   * @return The cluster recommendation response.
+   */
+  public CompletableFuture<ClusterRecommendationResponse> getClusterRecommendationsAsync(
+      String preferredClusterId,
+      String requiredGeo);
 }
diff --git a/java/src/main/java/com/microsoft/tunnels/management/TunnelClusterSource.java b/java/src/main/java/com/microsoft/tunnels/management/TunnelClusterSource.java
new file mode 100644
index 00000000..d8349927
--- /dev/null
+++ b/java/src/main/java/com/microsoft/tunnels/management/TunnelClusterSource.java
@@ -0,0 +1,78 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+package com.microsoft.tunnels.management;
+
+/**
+ * How the cluster for a tunnel create request was chosen.
+ *
+ * <p>When a create request does not specify a cluster, the client asks the recommendations
+ * API which cluster to use. That call can fail, and when it does the client falls back to
+ * global (Traffic Manager) routing, which still works but picks the nearest cluster by
+ * latency rather than the recommended one. The fallback is therefore invisible to the
+ * caller, so this value records which path was actually taken. It is reported to the
+ * service on every create request via the {@code X-Tunnel-Cluster-Source} header.</p>
+ *
+ * <p>Matches the values used by the C#, Go, and TypeScript SDKs.</p>
+ */
+public enum TunnelClusterSource {
+  /**
+   * The caller specified the cluster, so no recommendation was requested.
+   */
+  EXPLICIT("explicit"),
+
+  /**
+   * The recommendations API was called and its cluster was used.
+   */
+  RECOMMENDED("recommended"),
+
+  /**
+   * The recommendations API rejected the caller's token, and the retry without a token
+   * succeeded. Routing is correct but the caller was not identified, so it is treated as
+   * anonymous and cannot be assigned a service tier. This indicates a token problem on the
+   * caller's side that would otherwise be invisible.
+   */
+  RECOMMENDED_AFTER_AUTH_REJECTED("recommended-after-auth-rejected"),
+
+  /**
+   * The recommendations API returned unauthorized even without a token, so global routing
+   * was used instead.
+   */
+  FALLBACK_AUTH_FAILED("fallback-auth-failed"),
+
+  /**
+   * The recommendations API returned no cluster, so global routing was used instead.
+   */
+  FALLBACK_EMPTY("fallback-empty"),
+
+  /**
+   * The recommendations API call failed, so global routing was used instead.
+   */
+  FALLBACK_ERROR("fallback-error");
+
+  private final String headerValue;
+
+  TunnelClusterSource(String headerValue) {
+    this.headerValue = headerValue;
+  }
+
+  /**
+   * Gets the stable wire value sent to the service in the {@code X-Tunnel-Cluster-Source}
+   * header, which is what makes the client-side selection path visible in service telemetry.
+   *
+   * @return The header value for this source.
+   */
+  public String toHeaderValue() {
+    return this.headerValue;
+  }
+
+  /**
+   * Gets a value indicating whether this source means the recommendations API was bypassed
+   * or failed, so the tunnel was placed by global routing rather than by recommendation.
+   *
+   * @return True if this source is a fallback source.
+   */
+  public boolean isFallback() {
+    return this == FALLBACK_AUTH_FAILED || this == FALLBACK_EMPTY || this == FALLBACK_ERROR;
+  }
+}
diff --git a/java/src/main/java/com/microsoft/tunnels/management/TunnelManagementClient.java b/java/src/main/java/com/microsoft/tunnels/management/TunnelManagementClient.java
index 9c8c7f9d..4267a4d1 100644
--- a/java/src/main/java/com/microsoft/tunnels/management/TunnelManagementClient.java
+++ b/java/src/main/java/com/microsoft/tunnels/management/TunnelManagementClient.java
@@ -6,6 +6,7 @@
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import com.microsoft.tunnels.contracts.ClusterDetails;
+import com.microsoft.tunnels.contracts.ClusterRecommendationResponse;
 import com.microsoft.tunnels.contracts.NamedRateStatus;
 import com.microsoft.tunnels.contracts.Tunnel;
 import com.microsoft.tunnels.contracts.TunnelAccessControlEntry;
@@ -31,6 +32,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -57,8 +60,10 @@ public class TunnelManagementClient implements ITunnelManagementClient {
   private static final String endpointsApiSubPath = "/endpoints";
   private static final String portsApiSubPath = "/ports";
   private String clustersApiPath = "/clusters";
+  private static final String recommendationsApiSubPath = "/recommendations";
   private static final String tunnelAuthenticationScheme = "Tunnel";
   private static final String checkTunnelNamePath = ":checkNameAvailability";
+  private static final String clusterSourceHeaderName = "X-Tunnel-Cluster-Source";
   private static final int CreateNameRetries = 3;
 
   // Access Scopes
@@ -171,21 +176,34 @@ private <T, U> CompletableFuture<U> requestAsync(
       T requestObject,
       Type responseType) {
     return createHttpRequest(tunnel, options, requestMethod, uri, requestObject, scopes)
-      .thenCompose(request -> {
-        long startTime = System.nanoTime();
-        return this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
-          .thenApply(response -> {
-            long stopTime = System.nanoTime();
-            long durationMs = (stopTime - startTime) / 1000000;
-            var statusCode = response.statusCode();
-            var message = requestMethod + " " + uri + " -> " + statusCode + " (" + durationMs + " ms)";
-            if (statusCode >= 200 && statusCode < 300) {
-              logger.info(message);
-            } else {
-              logger.warn(message);
-            }
-            return response;
-          });
+      .thenCompose(request -> sendAndParseAsync(request, requestMethod, uri, responseType));
+  }
+
+  /**
+   * Sends an already-built request and parses the response, logging the outcome.
+   *
+   * <p>Split out of {@link #requestAsync} so that callers that build their own request (such
+   * as cluster recommendations, which needs to send an explicit auth override and retry with
+   * a different one) can reuse the same send/log/parse logic.</p>
+   */
+  private <U> CompletableFuture<U> sendAndParseAsync(
+      HttpRequest request,
+      HttpMethod requestMethod,
+      URI uri,
+      Type responseType) {
+    long startTime = System.nanoTime();
+    return this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
+      .thenApply(response -> {
+        long stopTime = System.nanoTime();
+        long durationMs = (stopTime - startTime) / 1000000;
+        var statusCode = response.statusCode();
+        var message = requestMethod + " " + uri + " -> " + statusCode + " (" + durationMs + " ms)";
+        if (statusCode >= 200 && statusCode < 300) {
+          logger.info(message);
+        } else {
+          logger.warn(message);
+        }
+        return response;
       })
       .thenApply(response -> parseResponse(response, responseType));
   }
@@ -243,35 +261,54 @@ private <T> CompletableFuture<HttpRequest> createHttpRequest(
       URI uri,
       T requestObject,
       String[] accessTokenScopes) {
-    return getAuthHeaderValue(tunnel, options, accessTokenScopes).thenApply(authHeaderValue -> {
-      String userAgentString = "";
-      for (ProductHeaderValue userAgent : this.userAgents) {
-        userAgentString = userAgent.productName
-            + "/" + userAgent.version + " " + userAgentString;
-      }
-      userAgentString = userAgentString + SDK_USER_AGENT;
-      var requestBuilder = HttpRequest.newBuilder()
-          .uri(uri)
-          .header(USER_AGENT_HEADER, userAgentString)
-          .header(CONTENT_TYPE_HEADER, "application/json");
-      if (StringUtils.isNotBlank(authHeaderValue)) {
-        requestBuilder.header(AUTH_HEADER, authHeaderValue);
-      }
+    return getAuthHeaderValue(tunnel, options, accessTokenScopes).thenApply(authHeaderValue ->
+        buildHttpRequest(options, requestMethod, uri, requestObject, authHeaderValue));
+  }
 
-      if (options != null && options.additionalHeaders != null) {
-        options.additionalHeaders.forEach(
-          (key, value) -> requestBuilder.header(key, value)
-        );
-      }
+  /**
+   * Builds an HTTP request using an already-resolved authorization header value.
+   *
+   * <p>Split out of {@link #createHttpRequest} so that callers that need to resolve
+   * authorization themselves (such as cluster recommendations, which sends an explicit token
+   * and retries once without it if rejected) can build a request without going through
+   * {@link #getAuthHeaderValue}'s tunnel/options-based resolution.</p>
+   *
+   * @param authHeaderValue The resolved value for the Authorization header, or null/blank to
+   *                        send the request unauthenticated.
+   */
+  private <T> HttpRequest buildHttpRequest(
+      TunnelRequestOptions options,
+      HttpMethod requestMethod,
+      URI uri,
+      T requestObject,
+      String authHeaderValue) {
+    String userAgentString = "";
+    for (ProductHeaderValue userAgent : this.userAgents) {
+      userAgentString = userAgent.productName
+          + "/" + userAgent.version + " " + userAgentString;
+    }
+    userAgentString = userAgentString + SDK_USER_AGENT;
+    var requestBuilder = HttpRequest.newBuilder()
+        .uri(uri)
+        .header(USER_AGENT_HEADER, userAgentString)
+        .header(CONTENT_TYPE_HEADER, "application/json");
+    if (StringUtils.isNotBlank(authHeaderValue)) {
+      requestBuilder.header(AUTH_HEADER, authHeaderValue);
+    }
 
-      Gson gson = TunnelContracts.getGson();
-      var requestJson = gson.toJson(requestObject);
-      var bodyPublisher = requestMethod == HttpMethod.POST || requestMethod == HttpMethod.PUT
-          ? BodyPublishers.ofString(requestJson)
-          : BodyPublishers.noBody();
-      requestBuilder.method(requestMethod.toString(), bodyPublisher);
-      return requestBuilder.build();
-    });
+    if (options != null && options.additionalHeaders != null) {
+      options.additionalHeaders.forEach(
+        (key, value) -> requestBuilder.header(key, value)
+      );
+    }
+
+    Gson gson = TunnelContracts.getGson();
+    var requestJson = gson.toJson(requestObject);
+    var bodyPublisher = requestMethod == HttpMethod.POST || requestMethod == HttpMethod.PUT
+        ? BodyPublishers.ofString(requestJson)
+        : BodyPublishers.noBody();
+    requestBuilder.method(requestMethod.toString(), bodyPublisher);
+    return requestBuilder.build();
   }
 
   private <T> T parseResponse(HttpResponse<String> response, Type typeOfT) {
@@ -413,43 +450,143 @@ public CompletableFuture<Tunnel> createTunnelAsync(Tunnel tunnel, TunnelRequestO
       tunnel.tunnelId = IdGeneration.generateTunnelId();
     }
 
-    options = options == null ? new TunnelRequestOptions() : options;
-    options.additionalHeaders = options.additionalHeaders == null
-        ? new HashMap<>() : options.additionalHeaders;
-    options.additionalHeaders.put("If-Not-Match", "*");
-
-    var uri = buildUri(tunnel, options, true);
-    final Type responseType = new TypeToken<Tunnel>() {
-    }.getType();
-    for (int i = 0; i <= CreateNameRetries; i++){
-      try {
-        return requestAsync(
-          tunnel,
-          options,
-          HttpMethod.PUT,
-          uri,
-          ManageAccessTokenScope,
-          convertTunnelForRequest(tunnel),
-          responseType);
-      }
-      catch (Exception e) {
-        if (generatedId) {
-          tunnel.tunnelId = IdGeneration.generateTunnelId();;
+    // Never mutate the caller's options (or its additionalHeaders map): create needs to add
+    // its own headers, including one that must not be spoofable by the caller.
+    var requestOptions = copyRequestOptionsForCreate(options);
+
+    // If the caller didn't specify a cluster, auto-select one via the recommendations API.
+    // Failures fall back to global routing.
+    CompletableFuture<TunnelClusterSource> clusterSourceFuture =
+        StringUtils.isBlank(tunnel.clusterId)
+            ? selectClusterAsync(tunnel, requestOptions.requiredGeo)
+            : CompletableFuture.completedFuture(TunnelClusterSource.EXPLICIT);
+
+    return clusterSourceFuture.thenCompose(clusterSource -> {
+      // Report the client-side selection path to the service. Recommendation fallbacks are
+      // otherwise invisible in service telemetry: a create that fell back looks identical to
+      // one that was never recommended at all.
+      requestOptions.additionalHeaders.put(
+          clusterSourceHeaderName, clusterSource.toHeaderValue());
+
+      var uri = buildUri(tunnel, requestOptions, true);
+      final Type responseType = new TypeToken<Tunnel>() {
+      }.getType();
+      for (int i = 0; i <= CreateNameRetries; i++){
+        try {
+          return requestAsync(
+            tunnel,
+            requestOptions,
+            HttpMethod.PUT,
+            uri,
+            ManageAccessTokenScope,
+            convertTunnelForRequest(tunnel),
+            responseType);
         }
-        else{
-          throw e;
+        catch (Exception e) {
+          if (generatedId) {
+            tunnel.tunnelId = IdGeneration.generateTunnelId();;
+          }
+          else{
+            throw e;
+          }
         }
       }
+
+      return requestAsync(
+            tunnel,
+            requestOptions,
+            HttpMethod.PUT,
+            uri,
+            ManageAccessTokenScope,
+            convertTunnelForRequest(tunnel),
+            responseType);
+    });
+  }
+
+  /**
+   * Makes a copy of the given request options suitable for a create request, without
+   * mutating the caller's instance.
+   *
+   * <p>The copy gets the "If-Not-Match" header that tunnel creation has always sent
+   * (preserving the existing header name as-is; it does not match the standard
+   * "If-None-Match" precondition header, but fixing that is a separate, unrelated behavior
+   * change). Any caller-supplied {@code X-Tunnel-Cluster-Source} header is removed
+   * case-insensitively so it cannot be spoofed — the client always sets that header itself,
+   * to report how the cluster was actually selected.</p>
+   */
+  private TunnelRequestOptions copyRequestOptionsForCreate(TunnelRequestOptions options) {
+    var copy = copyRequestOptions(options);
+    copy.additionalHeaders.put("If-Not-Match", "*");
+    return copy;
+  }
+
+  /**
+   * Makes a shallow copy of the given request options (or a fresh default instance if null),
+   * with a new, mutable {@code additionalHeaders} map that excludes any caller-supplied
+   * {@code X-Tunnel-Cluster-Source} header (case-insensitively), so a copy is always safe to
+   * mutate without affecting the caller or letting the caller spoof that header.
+   */
+  private TunnelRequestOptions copyRequestOptions(TunnelRequestOptions options) {
+    var copy = new TunnelRequestOptions();
+    if (options != null) {
+      copy.accessToken = options.accessToken;
+      copy.additionalQueryParameters = options.additionalQueryParameters == null
+          ? null : new HashMap<>(options.additionalQueryParameters);
+      copy.followRedirects = options.followRedirects;
+      copy.includePorts = options.includePorts;
+      copy.includeAccessControl = options.includeAccessControl;
+      copy.labels = options.labels;
+      copy.requireAllLabels = options.requireAllLabels;
+      copy.tokenScopes = options.tokenScopes;
+      copy.forceRename = options.forceRename;
+      copy.limit = options.limit;
+      copy.requiredGeo = options.requiredGeo;
     }
 
-    return requestAsync(
-          tunnel,
-          options,
-          HttpMethod.PUT,
-          uri,
-          ManageAccessTokenScope,
-          convertTunnelForRequest(tunnel),
-          responseType);
+    copy.additionalHeaders = new HashMap<>();
+    if (options != null && options.additionalHeaders != null) {
+      options.additionalHeaders.forEach((key, value) -> {
+        if (!clusterSourceHeaderName.equalsIgnoreCase(key)) {
+          copy.additionalHeaders.put(key, value);
+        }
+      });
+    }
+    return copy;
+  }
+
+  /**
+   * Attempts to auto-select a cluster for a new tunnel via the recommendations API, setting
+   * {@code tunnel.clusterId} when a usable recommendation comes back.
+   *
+   * <p>Failures fall back to global (Traffic Manager) routing, which still succeeds but
+   * picks the nearest cluster by latency rather than the recommended one — a silent fallback
+   * that would otherwise look identical to normal operation. This never fails: the outcome is
+   * instead reported via the returned {@link TunnelClusterSource}, which becomes the create
+   * request's {@code X-Tunnel-Cluster-Source} header.</p>
+   */
+  private CompletableFuture<TunnelClusterSource> selectClusterAsync(
+      Tunnel tunnel, String requiredGeo) {
+    return getClusterRecommendationsInternalAsync(null /* preferredClusterId */, requiredGeo)
+        .handle((result, throwable) -> {
+          if (throwable != null) {
+            Throwable cause = unwrapAsyncException(throwable);
+            TunnelClusterSource source = cause instanceof HttpResponseException
+                && isUnauthorizedOrForbidden((HttpResponseException) cause)
+                ? TunnelClusterSource.FALLBACK_AUTH_FAILED
+                : TunnelClusterSource.FALLBACK_ERROR;
+            return source;
+          }
+
+          if (result.response != null
+              && StringUtils.isNotBlank(result.response.recommendedClusterId)) {
+            tunnel.clusterId = result.response.recommendedClusterId;
+            return result.authRejected
+                ? TunnelClusterSource.RECOMMENDED_AFTER_AUTH_REJECTED
+                : TunnelClusterSource.RECOMMENDED;
+          }
+
+          return TunnelClusterSource.FALLBACK_EMPTY;
+        });
   }
 
   private Tunnel convertTunnelForRequest(Tunnel tunnel) {
@@ -959,4 +1096,150 @@ public CompletableFuture<Collection<NamedRateStatus>> listUserLimitsAsync() {
       throw new Error("Error parsing URI: " + this.baseAddress + TunnelManagementClient.userLimitsApiPath);
     }
   }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public CompletableFuture<ClusterRecommendationResponse> getClusterRecommendationsAsync(
+      String preferredClusterId,
+      String requiredGeo) {
+    return getClusterRecommendationsInternalAsync(preferredClusterId, requiredGeo)
+        .thenApply(result -> result.response);
+  }
+
+  /**
+   * Requests cluster recommendations, reporting whether the caller's token was rejected.
+   *
+   * <p>The token from the user token callback is resolved once and sent explicitly on the
+   * request (rather than through {@link #getAuthHeaderValue}/{@link #createHttpRequest}, so
+   * that it can be retried with a different, explicit auth override). This is so the service
+   * can identify the caller and apply their service tier. If the token is rejected with a 401
+   * or 403 the request is retried once without it, because the service rejects a bad token
+   * before the controller runs and does not fall back to treating the caller as anonymous.
+   * Without the retry, one expired token would silently disable recommendation-based routing
+   * for that caller. A response with no token to offer is an ordinary anonymous request, not
+   * a rejected one, so it is not retried. Failures other than a 401/403 on the authenticated
+   * attempt (500s, network errors, parse failures) are also not retried.</p>
+   *
+   * <p>Uses {@code handle} plus {@code thenCompose} rather than {@code exceptionallyCompose}
+   * (Java 12+) to stay on the Java 11 API surface this project targets.</p>
+   */
+  private CompletableFuture<ClusterRecommendationResult> getClusterRecommendationsInternalAsync(
+      String preferredClusterId,
+      String requiredGeo) {
+    var uri = buildClusterRecommendationsUri(preferredClusterId, requiredGeo);
+    final Type responseType = new TypeToken<ClusterRecommendationResponse>() {
+    }.getType();
+
+    CompletableFuture<String> userTokenFuture;
+    try {
+      userTokenFuture = this.userTokenCallback.get();
+    } catch (RuntimeException exception) {
+      userTokenFuture = new CompletableFuture<>();
+      userTokenFuture.completeExceptionally(exception);
+    }
+
+    return userTokenFuture.thenCompose(userAuthHeader -> {
+      if (StringUtils.isBlank(userAuthHeader)) {
+        // No token to offer, so this is an ordinary anonymous request rather than a
+        // rejected one.
+        var request = buildHttpRequest(null, HttpMethod.GET, uri, null, null);
+        CompletableFuture<ClusterRecommendationResponse> response =
+            sendAndParseAsync(request, HttpMethod.GET, uri, responseType);
+        return response.thenApply(r -> new ClusterRecommendationResult(r, false));
+      }
+
+      var authedRequest = buildHttpRequest(null, HttpMethod.GET, uri, null, userAuthHeader);
+      CompletableFuture<ClusterRecommendationResponse> authedResponse =
+          sendAndParseAsync(authedRequest, HttpMethod.GET, uri, responseType);
+
+      CompletableFuture<CompletableFuture<ClusterRecommendationResult>> handled =
+          authedResponse.handle((response, throwable) -> {
+            if (throwable == null) {
+              return CompletableFuture.completedFuture(
+                  new ClusterRecommendationResult(response, false));
+            }
+
+            Throwable cause = unwrapAsyncException(throwable);
+            if (cause instanceof HttpResponseException
+                && isUnauthorizedOrForbidden((HttpResponseException) cause)) {
+              // The service rejects a bad token before the controller runs rather than
+              // falling back to treating the caller as anonymous, so retry without it.
+              var anonymousRequest = buildHttpRequest(null, HttpMethod.GET, uri, null, null);
+              CompletableFuture<ClusterRecommendationResponse> anonymousResponse =
+                  sendAndParseAsync(anonymousRequest, HttpMethod.GET, uri, responseType);
+              return anonymousResponse.thenApply(r -> new ClusterRecommendationResult(r, true));
+            }
+
+            // 500s, network errors, and parse failures are not retried: only an actual
+            // auth rejection justifies falling back to an anonymous attempt.
+            CompletableFuture<ClusterRecommendationResult> failed = new CompletableFuture<>();
+            failed.completeExceptionally(cause);
+            return failed;
+          });
+      return handled.thenCompose(future -> future);
+    });
+  }
+
+  private URI buildClusterRecommendationsUri(String preferredClusterId, String requiredGeo) {
+    var queryParts = new ArrayList<String>();
+    if (StringUtils.isNotBlank(preferredClusterId)) {
+      queryParts.add("preferredClusterId=" + urlEncode(preferredClusterId));
+    }
+    if (StringUtils.isNotBlank(requiredGeo)) {
+      queryParts.add("requiredGeo=" + urlEncode(requiredGeo));
+    }
+    var query = queryParts.isEmpty() ? null : String.join("&", queryParts);
+    return buildUri(
+        null /* clusterId */, clustersApiPath + recommendationsApiSubPath, null /* options */,
+        query);
+  }
+
+  private static String urlEncode(String value) {
+    try {
+      return URLEncoder.encode(value, "UTF-8");
+    } catch (UnsupportedEncodingException e) {
+      throw new Error("Error encoding value: " + value);
+    }
+  }
+
+  /**
+   * Reports whether the given exception is an {@link HttpResponseException} carrying a 401
+   * or 403 status code from the service.
+   */
+  private static boolean isUnauthorizedOrForbidden(HttpResponseException exception) {
+    return exception.statusCode == 401 || exception.statusCode == 403;
+  }
+
+  /**
+   * Unwraps {@link CompletionException} and {@link ExecutionException} wrappers that
+   * {@link CompletableFuture} chains add around the exception that actually caused a stage
+   * to fail, so callers can inspect the real cause (for example an
+   * {@link HttpResponseException} status code) regardless of how many async stages it
+   * propagated through.
+   */
+  private static Throwable unwrapAsyncException(Throwable throwable) {
+    Throwable current = throwable;
+    while ((current instanceof CompletionException || current instanceof ExecutionException)
+        && current.getCause() != null) {
+      current = current.getCause();
+    }
+    return current;
+  }
+
+  /**
+   * Result of an internal cluster-recommendations request, additionally reporting whether
+   * the caller's token was rejected and a retry without it was needed.
+   */
+  private static final class ClusterRecommendationResult {
+    private final ClusterRecommendationResponse response;
+    private final boolean authRejected;
+
+    private ClusterRecommendationResult(
+        ClusterRecommendationResponse response, boolean authRejected) {
+      this.response = response;
+      this.authRejected = authRejected;
+    }
+  }
 }
diff --git a/java/src/main/java/com/microsoft/tunnels/management/TunnelRequestOptions.java b/java/src/main/java/com/microsoft/tunnels/management/TunnelRequestOptions.java
index e1b58ec0..39b895ae 100644
--- a/java/src/main/java/com/microsoft/tunnels/management/TunnelRequestOptions.java
+++ b/java/src/main/java/com/microsoft/tunnels/management/TunnelRequestOptions.java
@@ -119,6 +119,19 @@ public class TunnelRequestOptions {
    */
   public Integer limit;
 
+  /**
+   * Gets or sets an optional Azure geography filter used when a cluster is automatically
+   * recommended during tunnel creation.
+   *
+   * <p>This option only applies to {@code TunnelManagementClient.createTunnelAsync} when the
+   * tunnel does not already specify a {@code clusterId}. In that case the value is forwarded
+   * to the cluster recommendations request so that only clusters in the specified geo are
+   * eligible for automatic selection. It has no effect when a cluster is explicitly set or on
+   * any other request, and it is not sent as part of the create-tunnel request itself (so it
+   * is intentionally not included in {@link #toQueryString()}). Added for parity with the
+   * C#, Go, and TypeScript SDKs, which already forward this option the same way.</p>
+   */
+  public String requiredGeo;
 
   /**
    * Converts tunnel request options to a query string for HTTP requests to the
diff --git a/java/src/test/java/com/microsoft/tunnels/TunnelClusterRecommendationsTests.java b/java/src/test/java/com/microsoft/tunnels/TunnelClusterRecommendationsTests.java
new file mode 100644
index 00000000..041ba2a5
--- /dev/null
+++ b/java/src/test/java/com/microsoft/tunnels/TunnelClusterRecommendationsTests.java
@@ -0,0 +1,556 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+package com.microsoft.tunnels;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.microsoft.tunnels.contracts.Tunnel;
+import com.microsoft.tunnels.management.ProductHeaderValue;
+import com.microsoft.tunnels.management.TunnelManagementClient;
+import com.microsoft.tunnels.management.TunnelRequestOptions;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Deterministic, non-live tests for the cluster-recommendation handling in
+ * {@link TunnelManagementClient#createTunnelAsync}.
+ *
+ * <p>These tests run against a local, in-process HTTP server rather than the live tunnel
+ * service (deliberately not extending {@link TunnelTest}, which targets the real service),
+ * so they can assert exact request counts, headers, and retry behavior without any network
+ * dependency.</p>
+ */
+public class TunnelClusterRecommendationsTests {
+  private static final String API_VERSION = "2023-09-27-preview";
+  private static final String CLUSTER_SOURCE_HEADER = "X-Tunnel-Cluster-Source";
+  private static final String RECOMMENDATIONS_PATH = "/clusters/recommendations";
+
+  private HttpServer server;
+  private List<CapturedRequest> requests;
+  private Function<CapturedRequest, TestResponse> handler;
+
+  @Before
+  public void startServer() throws IOException {
+    this.requests = Collections.synchronizedList(new ArrayList<>());
+    // Port 0 lets the OS pick a free local port so tests never collide.
+    this.server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+    this.server.createContext("/", this::handleExchange);
+    this.server.setExecutor(null);
+    this.server.start();
+  }
+
+  @After
+  public void stopServer() {
+    if (this.server != null) {
+      this.server.stop(0);
+    }
+  }
+
+  private String baseAddress() {
+    return "http://localhost:" + this.server.getAddress().getPort();
+  }
+
+  private void handleExchange(HttpExchange exchange) throws IOException {
+    try {
+      var body = readBody(exchange.getRequestBody());
+      var capturedHeaders = new TreeMap<String, String>(String.CASE_INSENSITIVE_ORDER);
+      exchange.getRequestHeaders().forEach((name, values) -> {
+        if (!values.isEmpty()) {
+          capturedHeaders.put(name, values.get(0));
+        }
+      });
+      var request = new CapturedRequest(
+          exchange.getRequestMethod(),
+          exchange.getRequestURI().toString(),
+          capturedHeaders,
+          body);
+      this.requests.add(request);
+
+      var response = this.handler.apply(request);
+      byte[] payload = response.body.getBytes(StandardCharsets.UTF_8);
+
+      // Explicit Content-Length (never chunked) and Connection: close keep each request/
+      // response exchange self-contained and simple to reason about in a short-lived,
+      // per-test local server. sendResponseHeaders sets Content-Length itself from the
+      // length argument below, so it must not also be set manually here.
+      exchange.getResponseHeaders().set("Content-Type", "application/json");
+      exchange.getResponseHeaders().set("Connection", "close");
+      exchange.sendResponseHeaders(response.statusCode, payload.length);
+      try (OutputStream responseBody = exchange.getResponseBody()) {
+        responseBody.write(payload);
+      }
+    } finally {
+      exchange.close();
+    }
+  }
+
+  private static String readBody(InputStream stream) throws IOException {
+    var buffer = new ByteArrayOutputStream();
+    var chunk = new byte[4096];
+    int read;
+    while ((read = stream.read(chunk)) != -1) {
+      buffer.write(chunk, 0, read);
+    }
+    return buffer.toString(StandardCharsets.UTF_8.name());
+  }
+
+  private TunnelManagementClient createClient(Supplier<CompletableFuture<String>> userTokenCallback) {
+    var userAgent = new ProductHeaderValue("cluster-recommendations-test", "1.0");
+    return new TunnelManagementClient(
+        new ProductHeaderValue[] { userAgent },
+        userTokenCallback,
+        baseAddress(),
+        API_VERSION);
+  }
+
+  private static Supplier<CompletableFuture<String>> tokenCallback(String bearerToken) {
+    return () -> CompletableFuture.completedFuture("Bearer " + bearerToken);
+  }
+
+  private static TestResponse recommendationResponse(String clusterId) {
+    var json = clusterId == null
+        ? "{\"recommendations\":[]}"
+        : "{\"recommendedClusterId\":\"" + clusterId + "\",\"recommendations\":[]}";
+    return new TestResponse(200, json);
+  }
+
+  private static TestResponse unauthorized(int statusCode) {
+    return new TestResponse(statusCode, "{\"error\":\"unauthorized\"}");
+  }
+
+  private static TestResponse createResponse(String tunnelId) {
+    return new TestResponse(200, "{\"tunnelId\":\"" + tunnelId + "\"}");
+  }
+
+  private List<CapturedRequest> recommendationRequests() {
+    var result = new ArrayList<CapturedRequest>();
+    for (var request : this.requests) {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        result.add(request);
+      }
+    }
+    return result;
+  }
+
+  private List<CapturedRequest> createRequests() {
+    var result = new ArrayList<CapturedRequest>();
+    for (var request : this.requests) {
+      if (request.method.equals("PUT")) {
+        result.add(request);
+      }
+    }
+    return result;
+  }
+
+  private Tunnel createTunnel(
+      TunnelManagementClient client, Tunnel tunnel, TunnelRequestOptions options) {
+    try {
+      return client.createTunnelAsync(tunnel, options).get(10, java.util.concurrent.TimeUnit.SECONDS);
+    } catch (java.util.concurrent.ExecutionException | InterruptedException
+        | java.util.concurrent.TimeoutException e) {
+      throw new AssertionError("createTunnelAsync failed unexpectedly: " + e.getCause(), e);
+    }
+  }
+
+  @Test
+  public void authenticatedRecommendation_SelectsClusterOnCreate() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        assertEquals("Bearer usertoken", request.headers.get("Authorization"));
+        return recommendationResponse("usw4");
+      }
+      assertEquals("recommended", request.headers.get(CLUSTER_SOURCE_HEADER));
+      return createResponse("tunnel001");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel001";
+
+    createTunnel(client, tunnel, null);
+
+    assertEquals(1, recommendationRequests().size());
+    assertEquals(1, createRequests().size());
+    assertEquals("usw4", tunnel.clusterId);
+    assertEquals("recommended", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void unauthorized401_RetriesAnonymouslyAndSucceeds() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        if (request.headers.containsKey("Authorization")) {
+          return unauthorized(401);
+        }
+        return recommendationResponse("usw4");
+      }
+      return createResponse("tunnel002");
+    };
+
+    var client = createClient(tokenCallback("expired"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel002";
+
+    createTunnel(client, tunnel, null);
+
+    var recommendationCalls = recommendationRequests();
+    assertEquals(2, recommendationCalls.size());
+    assertEquals("Bearer expired", recommendationCalls.get(0).headers.get("Authorization"));
+    // The retry must not carry any Authorization header at all -- not just a blank one.
+    assertFalse(recommendationCalls.get(1).headers.containsKey("Authorization"));
+    assertEquals("usw4", tunnel.clusterId);
+    assertEquals(
+        "recommended-after-auth-rejected",
+        createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void forbidden403_RetriesAnonymouslyAndSucceeds() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        if (request.headers.containsKey("Authorization")) {
+          return unauthorized(403);
+        }
+        return recommendationResponse("usw4");
+      }
+      return createResponse("tunnel003");
+    };
+
+    var client = createClient(tokenCallback("expired"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel003";
+
+    createTunnel(client, tunnel, null);
+
+    var recommendationCalls = recommendationRequests();
+    assertEquals(2, recommendationCalls.size());
+    assertFalse(recommendationCalls.get(1).headers.containsKey("Authorization"));
+    assertEquals("usw4", tunnel.clusterId);
+    assertEquals(
+        "recommended-after-auth-rejected",
+        createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void serverError500_DoesNotRetryAndFallsBack() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        return new TestResponse(500, "{\"error\":\"boom\"}");
+      }
+      return createResponse("tunnel004");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel004";
+
+    createTunnel(client, tunnel, null);
+
+    // A 500 is not an auth rejection, so it must not be retried: exactly one attempt.
+    assertEquals(1, recommendationRequests().size());
+    assertNull(tunnel.clusterId);
+    assertEquals("fallback-error", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void anonymousRetryAlsoUnauthorized_FallsBackToAuthFailed() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        return unauthorized(401);
+      }
+      return createResponse("tunnel005");
+    };
+
+    var client = createClient(tokenCallback("expired"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel005";
+
+    createTunnel(client, tunnel, null);
+
+    // One authenticated attempt, one anonymous retry, both rejected.
+    assertEquals(2, recommendationRequests().size());
+    assertNull(tunnel.clusterId);
+    assertEquals(
+        "fallback-auth-failed", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void emptyRecommendationResponse_FallsBackToEmpty() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        return recommendationResponse(null);
+      }
+      return createResponse("tunnel006");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel006";
+
+    createTunnel(client, tunnel, null);
+
+    // A successful, complete response with no cluster does not need a retry.
+    assertEquals(1, recommendationRequests().size());
+    assertNull(tunnel.clusterId);
+    assertEquals("fallback-empty", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void explicitClusterId_SkipsRecommendationsEntirely() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        fail("Recommendations should not be called when a cluster is already set.");
+      }
+      return createResponse("tunnel007");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel007";
+    tunnel.clusterId = "usw3";
+
+    createTunnel(client, tunnel, null);
+
+    assertEquals(0, recommendationRequests().size());
+    assertEquals(1, createRequests().size());
+    assertEquals("explicit", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void allClusterSourceHeaderValues_MatchExpectedWireValues() {
+    // Reconfirms the exact wire values in one place, so a rename of the enum constants can't
+    // silently drift from what the service expects.
+    var expected = new HashSet<>(java.util.Arrays.asList(
+        "explicit",
+        "recommended",
+        "recommended-after-auth-rejected",
+        "fallback-auth-failed",
+        "fallback-empty",
+        "fallback-error"));
+
+    var actual = new HashSet<String>();
+    for (var source : com.microsoft.tunnels.management.TunnelClusterSource.values()) {
+      actual.add(source.toHeaderValue());
+    }
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void callerOptions_AreNotMutatedAndSourceHeaderCannotBeSpoofed() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        fail("Recommendations should not be called when a cluster is already set.");
+      }
+      return createResponse("tunnel008");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel008";
+    tunnel.clusterId = "usw3";
+
+    var options = new TunnelRequestOptions();
+    var callerHeaders = new HashMap<String, String>();
+    callerHeaders.put("X-Tunnel-Cluster-Source", "explicit-spoofed");
+    callerHeaders.put("X-Custom-Header", "keep-me");
+    options.additionalHeaders = callerHeaders;
+
+    createTunnel(client, tunnel, options);
+
+    var sentHeaders = createRequests().get(0).headers;
+    // The client's own value wins over anything the caller supplied.
+    assertEquals("explicit", sentHeaders.get(CLUSTER_SOURCE_HEADER));
+    // Other caller headers are preserved.
+    assertEquals("keep-me", sentHeaders.get("X-Custom-Header"));
+
+    // The caller's options object (and its additionalHeaders map) must be untouched: no
+    // "If-Not-Match" header added, and the spoofed value still present exactly as given.
+    assertEquals(2, options.additionalHeaders.size());
+    assertEquals("explicit-spoofed", options.additionalHeaders.get("X-Tunnel-Cluster-Source"));
+    assertEquals("keep-me", options.additionalHeaders.get("X-Custom-Header"));
+    assertFalse(options.additionalHeaders.containsKey("If-Not-Match"));
+  }
+
+  @Test
+  public void callerOptions_SpoofHeaderRemovedCaseInsensitively() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        return recommendationResponse("usw4");
+      }
+      return createResponse("tunnel009");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel009";
+
+    var options = new TunnelRequestOptions();
+    var callerHeaders = new HashMap<String, String>();
+    // Lowercase variant of the header name the client uses -- must still be treated as a
+    // spoof attempt and removed, since HTTP header names are case-insensitive.
+    callerHeaders.put("x-tunnel-cluster-source", "recommended-spoofed");
+    options.additionalHeaders = callerHeaders;
+
+    createTunnel(client, tunnel, options);
+
+    var sentHeaders = createRequests().get(0).headers;
+    assertEquals("recommended", sentHeaders.get(CLUSTER_SOURCE_HEADER));
+    assertTrue(sentHeaders.get(CLUSTER_SOURCE_HEADER) == null
+        || !sentHeaders.get(CLUSTER_SOURCE_HEADER).contains("spoofed"));
+
+    // Caller's map is untouched.
+    assertEquals(1, options.additionalHeaders.size());
+    assertEquals("recommended-spoofed", options.additionalHeaders.get("x-tunnel-cluster-source"));
+  }
+
+  @Test
+  public void unauthenticatedClient_SendsNoAuthorizationHeaderAndDoesNotRetry() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        assertFalse(request.headers.containsKey("Authorization"));
+        return recommendationResponse("usw4");
+      }
+      return createResponse("tunnel010");
+    };
+
+    // No user token callback configured at all.
+    var client = createClient(null);
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel010";
+
+    createTunnel(client, tunnel, null);
+
+    assertEquals(1, recommendationRequests().size());
+    assertEquals("usw4", tunnel.clusterId);
+    assertEquals("recommended", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  @Test
+  public void getClusterRecommendationsAsync_ReturnsResponseDirectly() {
+    this.handler = request -> {
+      assertTrue(request.path.startsWith(RECOMMENDATIONS_PATH));
+      assertTrue(request.path.contains("preferredClusterId=preferred+cluster"));
+      assertTrue(request.path.contains("requiredGeo=eu+west"));
+      return recommendationResponse("usw5");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var response = client
+        .getClusterRecommendationsAsync("preferred cluster", "eu west")
+        .join();
+
+    assertEquals("usw5", response.recommendedClusterId);
+    assertEquals(1, recommendationRequests().size());
+  }
+
+  @Test
+  public void requiredGeo_IsForwardedOnlyToRecommendations() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        assertTrue(request.path.contains("requiredGeo=eu+west"));
+        return recommendationResponse("usw4");
+      }
+      assertFalse(request.path.contains("requiredGeo"));
+      return createResponse("tunnel011");
+    };
+
+    var client = createClient(tokenCallback("usertoken"));
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel011";
+    var options = new TunnelRequestOptions();
+    options.requiredGeo = "eu west";
+
+    createTunnel(client, tunnel, options);
+
+    assertEquals("usw4", tunnel.clusterId);
+    assertEquals(1, recommendationRequests().size());
+    assertEquals(1, createRequests().size());
+  }
+
+  @Test
+  public void tokenCallbackFailure_FallsBackToCreate() {
+    this.handler = request -> {
+      if (request.path.startsWith(RECOMMENDATIONS_PATH)) {
+        fail("Recommendations request should not be sent when token resolution fails.");
+      }
+      return createResponse("tunnel012");
+    };
+
+    var callbackCount = new AtomicInteger();
+    Supplier<CompletableFuture<String>> tokenCallback = () -> {
+      if (callbackCount.getAndIncrement() == 0) {
+        throw new IllegalStateException("token unavailable");
+      }
+      return CompletableFuture.completedFuture("Bearer recovered");
+    };
+    var client = createClient(tokenCallback);
+    var tunnel = new Tunnel();
+    tunnel.tunnelId = "tunnel012";
+
+    createTunnel(client, tunnel, null);
+
+    assertEquals(2, callbackCount.get());
+    assertEquals(0, recommendationRequests().size());
+    assertEquals(
+        "fallback-error", createRequests().get(0).headers.get(CLUSTER_SOURCE_HEADER));
+  }
+
+  /**
+   * A recorded request as observed by the local server.
+   */
+  private static final class CapturedRequest {
+    private final String method;
+    private final String path;
+    private final TreeMap<String, String> headers;
+    private final String body;
+
+    private CapturedRequest(
+        String method, String path, TreeMap<String, String> headers, String body) {
+      this.method = method;
+      this.path = path;
+      this.headers = headers;
+      this.body = body;
+    }
+  }
+
+  /**
+   * A canned response for the local server to send back for a captured request.
+   */
+  private static final class TestResponse {
+    private final int statusCode;
+    private final String body;
+
+    private TestResponse(int statusCode, String body) {
+      this.statusCode = statusCode;
+      this.body = body;
+    }
+  }
+}