Add ClientOptions and HttpPipelinePolicyPosition Support to Schema Registry Builders (Azure#23057)

Author: alzimmermsft

sdk/schemaregistry/azure-data-schemaregistry/src/main/java/com/azure/data/schemaregistry/SchemaRegistryClientBuilder.java

@@ -6,10 +6,13 @@
 import com.azure.core.annotation.ServiceClientBuilder;
 import com.azure.core.credential.TokenCredential;
 import com.azure.core.http.HttpClient;
+import com.azure.core.http.HttpHeader;
 import com.azure.core.http.HttpHeaders;
 import com.azure.core.http.HttpPipeline;
 import com.azure.core.http.HttpPipelineBuilder;
+import com.azure.core.http.HttpPipelinePosition;
 import com.azure.core.http.policy.AddDatePolicy;
+import com.azure.core.http.policy.AddHeadersFromContextPolicy;
 import com.azure.core.http.policy.AddHeadersPolicy;
 import com.azure.core.http.policy.BearerTokenAuthenticationPolicy;
 import com.azure.core.http.policy.HttpLogDetailLevel;
@@ -20,11 +23,13 @@
 import com.azure.core.http.policy.RequestIdPolicy;
 import com.azure.core.http.policy.RetryPolicy;
 import com.azure.core.http.policy.UserAgentPolicy;
+import com.azure.core.util.ClientOptions;
 import com.azure.core.util.Configuration;
 import com.azure.core.util.CoreUtils;
 import com.azure.core.util.logging.ClientLogger;
 import com.azure.data.schemaregistry.implementation.AzureSchemaRegistry;
 import com.azure.data.schemaregistry.implementation.AzureSchemaRegistryBuilder;
+
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.time.temporal.ChronoUnit;
@@ -46,11 +51,15 @@ public class SchemaRegistryClientBuilder {
     private static final String CLIENT_PROPERTIES = "azure-data-schemaregistry-client.properties";
     private static final String NAME = "name";
     private static final String VERSION = "version";
-    private static final RetryPolicy DEFAULT_RETRY_POLICY =
-        new RetryPolicy("retry-after-ms", ChronoUnit.MILLIS);
+    private static final RetryPolicy DEFAULT_RETRY_POLICY = new RetryPolicy("retry-after-ms", ChronoUnit.MILLIS);
+    private static final AddHeadersPolicy API_HEADER_POLICY = new AddHeadersPolicy(new HttpHeaders()
+        .set("api-version", "2020-09-01-preview"));
 
     private final ConcurrentSkipListMap<String, Function<String, Object>> typeParserMap;
-    private final List<HttpPipelinePolicy> policies;
+
+    private final List<HttpPipelinePolicy> perCallPolicies = new ArrayList<>();
+    private final List<HttpPipelinePolicy> perRetryPolicies = new ArrayList<>();
+
     private final String clientName;
     private final String clientVersion;
 
@@ -59,26 +68,23 @@ public class SchemaRegistryClientBuilder {
     private HttpClient httpClient;
     private Integer maxSchemaMapSize;
     private TokenCredential credential;
+    private ClientOptions clientOptions;
     private HttpLogOptions httpLogOptions;
     private HttpPipeline httpPipeline;
     private RetryPolicy retryPolicy;
+    private Configuration configuration;
 
     /**
      * Constructor for CachedSchemaRegistryClientBuilder.  Supplies client defaults.
      */
     public SchemaRegistryClientBuilder() {
-        this.policies = new ArrayList<>();
         this.httpLogOptions = new HttpLogOptions();
         this.maxSchemaMapSize = null;
         this.typeParserMap = new ConcurrentSkipListMap<>(String.CASE_INSENSITIVE_ORDER);
         this.httpClient = null;
         this.credential = null;
         this.retryPolicy = new RetryPolicy("retry-after-ms", ChronoUnit.MILLIS);
 
-        HttpHeaders headers = new HttpHeaders();
-        headers.put("api-version", "2020-09-01-preview");
-        policies.add(new AddHeadersPolicy(headers));
-
         Map<String, String> properties = CoreUtils.getProperties(CLIENT_PROPERTIES);
         clientName = properties.getOrDefault(NAME, "UnknownName");
         clientVersion = properties.getOrDefault(VERSION, "UnknownVersion");
@@ -87,8 +93,8 @@ public SchemaRegistryClientBuilder() {
     /**
      * Sets the service endpoint for the Azure Schema Registry instance.
      *
-     * @return The updated {@link SchemaRegistryClientBuilder} object.
      * @param endpoint The URL of the Azure Schema Registry instance
+     * @return The updated {@link SchemaRegistryClientBuilder} object.
      * @throws NullPointerException if {@code endpoint} is null
      * @throws IllegalArgumentException if {@code endpoint} cannot be parsed into a valid URL
      */
@@ -159,8 +165,22 @@ public SchemaRegistryClientBuilder pipeline(HttpPipeline httpPipeline) {
     }
 
     /**
-     * Sets the {@link TokenCredential} to use when authenticating HTTP requests for this
-     * {@link SchemaRegistryAsyncClient}.
+     * Sets the configuration store that is used during construction of the service client.
+     *
+     * The default configuration store is a clone of the {@link Configuration#getGlobalConfiguration() global
+     * configuration store}, use {@link Configuration#NONE} to bypass using configuration settings during construction.
+     *
+     * @param configuration The configuration store used to
+     * @return The updated SchemaRegistryClientBuilder object.
+     */
+    public SchemaRegistryClientBuilder configuration(Configuration configuration) {
+        this.configuration = configuration;
+        return this;
+    }
+
+    /**
+     * Sets the {@link TokenCredential} to use when authenticating HTTP requests for this {@link
+     * SchemaRegistryAsyncClient}.
      *
      * @param credential {@link TokenCredential}
      * @return The updated {@link SchemaRegistryClientBuilder} object.
@@ -171,6 +191,22 @@ public SchemaRegistryClientBuilder credential(TokenCredential credential) {
         return this;
     }
 
+    /**
+     * Sets the {@link ClientOptions} which enables various options to be set on the client. For example setting an
+     * {@code applicationId} using {@link ClientOptions#setApplicationId(String)} to configure the {@link
+     * UserAgentPolicy} for telemetry/monitoring purposes.
+     *
+     * <p>More About <a href="https://azure.github.io/azure-sdk/general_azurecore.html#telemetry-policy">Azure Core:
+     * Telemetry policy</a>
+     *
+     * @param clientOptions {@link ClientOptions}.
+     * @return The updated SchemaRegistryClientBuilder object.
+     */
+    public SchemaRegistryClientBuilder clientOptions(ClientOptions clientOptions) {
+        this.clientOptions = clientOptions;
+        return this;
+    }
+
     /**
      * Sets the logging configuration for HTTP requests and responses.
      *
@@ -205,72 +241,96 @@ public SchemaRegistryClientBuilder retryPolicy(RetryPolicy retryPolicy) {
      * @throws NullPointerException If {@code policy} is {@code null}.
      */
     public SchemaRegistryClientBuilder addPolicy(HttpPipelinePolicy policy) {
-        policies.add(Objects.requireNonNull(policy, "'policy' cannot be null."));
+        Objects.requireNonNull(policy, "'policy' cannot be null.");
+
+        if (policy.getPipelinePosition() == HttpPipelinePosition.PER_CALL) {
+            perCallPolicies.add(policy);
+        } else {
+            perRetryPolicies.add(policy);
+        }
+
         return this;
     }
 
     /**
-     * Creates a {@link SchemaRegistryAsyncClient} based on options set in the builder.
-     * Every time {@code buildClient()} is called a new instance of {@link SchemaRegistryAsyncClient} is created.
+     * Creates a {@link SchemaRegistryAsyncClient} based on options set in the builder. Every time {@code buildClient()}
+     * is called a new instance of {@link SchemaRegistryAsyncClient} is created.
      *
      * If {@link #pipeline(HttpPipeline) pipeline} is set, then all HTTP pipeline related settings are ignored.
      *
      * @return A {@link SchemaRegistryAsyncClient} with the options set from the builder.
-     * @throws NullPointerException if {@link #endpoint(String) endpoint} and
-     * {@link #credential(TokenCredential) credential} are not set.
+     * @throws NullPointerException if {@link #endpoint(String) endpoint} and {@link #credential(TokenCredential)
+     * credential} are not set.
      */
     public SchemaRegistryAsyncClient buildAsyncClient() {
         Objects.requireNonNull(credential, "'credential' cannot be null");
         Objects.requireNonNull(endpoint, "'endpoint' cannot be null");
 
-        HttpPipeline pipeline = this.httpPipeline;
+        Configuration buildConfiguration = (configuration == null)
+            ? Configuration.getGlobalConfiguration()
+            : configuration;
+
+        HttpPipeline buildPipeline = this.httpPipeline;
         // Create a default Pipeline if it is not given
-        if (pipeline == null) {
+        if (buildPipeline == null) {
             // Closest to API goes first, closest to wire goes last.
             final List<HttpPipelinePolicy> policies = new ArrayList<>();
 
-            policies.add(new UserAgentPolicy(httpLogOptions.getApplicationId(), clientName, clientVersion,
-                Configuration.getGlobalConfiguration().clone()));
+            policies.add(new UserAgentPolicy(CoreUtils.getApplicationId(clientOptions, httpLogOptions), clientName,
+                clientVersion, buildConfiguration));
             policies.add(new RequestIdPolicy());
+            policies.add(new AddHeadersFromContextPolicy());
+            policies.add(API_HEADER_POLICY);
 
+            policies.addAll(perCallPolicies);
             HttpPolicyProviders.addBeforeRetryPolicies(policies);
 
             policies.add(retryPolicy == null ? DEFAULT_RETRY_POLICY : retryPolicy);
 
             policies.add(new AddDatePolicy());
-
             policies.add(new BearerTokenAuthenticationPolicy(credential, DEFAULT_SCOPE));
 
-            policies.addAll(this.policies);
+            policies.addAll(perRetryPolicies);
+
+            if (clientOptions != null) {
+                List<HttpHeader> clientOptionsHeaders = new ArrayList<>();
+                clientOptions.getHeaders()
+                    .forEach(header -> clientOptionsHeaders.add(new HttpHeader(header.getName(), header.getValue())));
+
+                if (!CoreUtils.isNullOrEmpty(clientOptionsHeaders)) {
+                    policies.add(new AddHeadersPolicy(new HttpHeaders(clientOptionsHeaders)));
+                }
+            }
+
             HttpPolicyProviders.addAfterRetryPolicies(policies);
 
             policies.add(new HttpLoggingPolicy(httpLogOptions));
 
-            pipeline = new HttpPipelineBuilder()
+            buildPipeline = new HttpPipelineBuilder()
                 .policies(policies.toArray(new HttpPipelinePolicy[0]))
                 .httpClient(httpClient)
+                .clientOptions(clientOptions)
                 .build();
         }
 
         AzureSchemaRegistry restService = new AzureSchemaRegistryBuilder()
             .endpoint(host)
-            .pipeline(pipeline)
+            .pipeline(buildPipeline)
             .buildClient();
 
-        this.maxSchemaMapSize = this.maxSchemaMapSize != null
-            ? this.maxSchemaMapSize
-            : SchemaRegistryAsyncClient.MAX_SCHEMA_MAP_SIZE_DEFAULT;
+        int buildMaxSchemaMapSize = (maxSchemaMapSize == null)
+            ? SchemaRegistryAsyncClient.MAX_SCHEMA_MAP_SIZE_DEFAULT
+            : maxSchemaMapSize;
 
-        return new SchemaRegistryAsyncClient(restService, maxSchemaMapSize, typeParserMap);
+        return new SchemaRegistryAsyncClient(restService, buildMaxSchemaMapSize, typeParserMap);
     }
 
     /**
-     * Creates synchronous {@link SchemaRegistryClient} instance.
-     * See async builder method for options validation.
+     * Creates synchronous {@link SchemaRegistryClient} instance. See async builder method for options validation.
      *
      * @return {@link SchemaRegistryClient} with the options set from the builder.
-     * @throws NullPointerException if {@link #endpoint(String) endpoint} and
-     * {@link #credential(TokenCredential) credential} are not set.
+     * @throws NullPointerException if {@link #endpoint(String) endpoint} and {@link #credential(TokenCredential)
+     * credential} are not set.
      */
     public SchemaRegistryClient buildClient() {
         return new SchemaRegistryClient(this.buildAsyncClient());