Removes apache http libraries and replace with azure core httpclient

CHANGELOG.md

@@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [6.0.0] - 2024-03-07
+### Changed
+- Replaced Apache CloseableHttpClient with configurable azure-core client.
+- (Breaking) HttpClientFactory now accepts clients implementing azure-core HttpClient.
+- (Breaking) HttpClientProperties and HttpClientPropertiesBuilder now use azure-core ProxyOptions.
+- Data client now wraps internal HTTP client.
+- Moved HTTP request tracing logic into a builder class.
+- Moved HTTP request building logic into a builder class.
+
 ## [5.0.5] - 2024-03-06
 ### Fixed
 - Fixed bugs in how ClientRequestProperties' servertimeout is set/get, and aligned the 3 different ways to set/get this option

README.md

@@ -59,7 +59,7 @@ And the SDK will know to use these values automatically.
 Alternatively, you can define a proxy programmatically when creating a client, using `HttpClientProperties`:  
 ```java
 HttpClientProperties httpClientProperties = HttpClientProperties.builder()
-        .proxy(new HttpHost("1.2.3.4", 8989))
+        .proxy(new ProxyOptions(ProxyOptions.Type.HTTP, new InetSocketAddress("myproxy.contoso.com", 8080)))
         .build();
 
 Client = ClientFactory.createClient(<engine_connection_string>, httpClientProperties); 

data/pom.xml

@@ -159,27 +159,17 @@
             <groupId>com.azure</groupId>
             <artifactId>azure-core</artifactId>
         </dependency>
-        <!-- https://mvnrepository.com/artifact/org.apache.httpcomponents/httpclient -->
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpclient</artifactId>
-            <version>${httpclient.version}</version>
+            <version>${apache.httpclient.version}</version>
             <exclusions>
-                <exclusion>
-                    <artifactId>httpcore</artifactId>
-                    <groupId>org.apache.httpcomponents</groupId>
-                </exclusion>
                 <exclusion>
                     <artifactId>commons-codec</artifactId>
                     <groupId>org.apache.commons</groupId>
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>org.apache.httpcomponents</groupId>
-            <artifactId>httpcore</artifactId>
-            <version>${httpcore.version}</version>
-        </dependency>
         <dependency>
             <groupId>org.jetbrains</groupId>
             <artifactId>annotations</artifactId>

data/src/main/java/com/microsoft/azure/kusto/data/BaseClient.java

@@ -0,0 +1,167 @@
+package com.microsoft.azure.kusto.data;
+
+import com.azure.core.http.*;
+import com.azure.core.util.Context;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.microsoft.azure.kusto.data.exceptions.*;
+import com.microsoft.azure.kusto.data.http.HttpRequestBuilder;
+import com.microsoft.azure.kusto.data.http.HttpStatus;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.util.Optional;
+
+public abstract class BaseClient implements Client, StreamingClient {
+
+    private static final int MAX_REDIRECT_COUNT = 1;
+
+    // Make logger available to implementations
+    protected static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+    private final HttpClient httpClient;
+
+    public BaseClient(HttpClient httpClient) {
+        this.httpClient = httpClient;
+    }
+
+    protected String post(HttpRequest request) throws DataServiceException {
+
+        // Todo: Add async version of this method
+
+        // Execute and get the response
+        try (HttpResponse response = httpClient.sendSync(request, Context.NONE)) {
+            String responseBody = Utils.isGzipResponse(response) ? Utils.gzipedInputToString(response.getBodyAsBinaryData().toStream())
+                    : response.getBodyAsBinaryData().toString();
+
+            if (responseBody != null) {
+                switch (response.getStatusCode()) {
+                    case HttpStatus.OK:
+                        return responseBody;
+                    case HttpStatus.TOO_MANY_REQS:
+                        throw new ThrottleException(request.getUrl().toString());
+                    default:
+                        throw createExceptionFromResponse(request.getUrl().toString(), response, null, responseBody);
+                }
+            }
+        }
+
+        return null;
+    }
+
+    protected InputStream postToStreamingOutput(HttpRequest request) throws DataServiceException {
+        return postToStreamingOutput(request, 0);
+    }
+
+    protected InputStream postToStreamingOutput(HttpRequest request, int redirectCount) throws DataServiceException {
+
+        boolean returnInputStream = false;
+        String errorFromResponse = null;
+
+        HttpResponse httpResponse = null;
+        try {
+
+            // Todo: Implement async version of this method
+            httpResponse = httpClient.sendSync(request, Context.NONE);
+
+            int responseStatusCode = httpResponse.getStatusCode();
+
+            if (responseStatusCode == HttpStatus.OK) {
+                returnInputStream = true;
+                return Utils.getResponseAsStream(httpResponse);
+            }
+
+            errorFromResponse = httpResponse.getBodyAsBinaryData().toString();
+            // Ideal to close here (as opposed to finally) so that if any data can't be flushed, the exception will be properly thrown and handled
+            httpResponse.close();
+
+            if (shouldPostToOriginalUrlDueToRedirect(redirectCount, responseStatusCode)) {
+                Optional<HttpHeader> redirectLocation = Optional.ofNullable(httpResponse.getHeaders().get(HttpHeaderName.LOCATION));
+                if (redirectLocation.isPresent() && !redirectLocation.get().getValue().equals(request.getUrl().toString())) {
+                    HttpRequest redirectRequest = HttpRequestBuilder
+                            .fromExistingRequest(request)
+                            .withURL(redirectLocation.get().getValue())
+                            .build();
+                    return postToStreamingOutput(redirectRequest, redirectCount + 1);
+                }
+            }
+        } catch (IOException ex) {
+            throw new DataServiceException(request.getUrl().toString(),
+                    "postToStreamingOutput failed to get or decompress response stream", ex, false);
+        } catch (Exception ex) {
+            throw createExceptionFromResponse(request.getUrl().toString(), httpResponse, ex, errorFromResponse);
+        } finally {
+            closeResourcesIfNeeded(returnInputStream, httpResponse);
+        }
+
+        throw createExceptionFromResponse(request.getUrl().toString(), httpResponse, null, errorFromResponse);
+    }
+
+    public static DataServiceException createExceptionFromResponse(String url, HttpResponse httpResponse, Exception thrownException, String errorFromResponse) {
+        if (httpResponse == null) {
+            return new DataServiceException(url, "POST failed to send request", thrownException, false);
+        }
+        /*
+         * TODO: When we add another streaming API that returns a KustoOperationResult, we'll need to handle the 2 types of content errors this API can return:
+         * (1) Inline error (engine identifies error after it starts building the json result), or (2) in the KustoOperationResult's QueryCompletionInformation,
+         * both of which present with "200 OK". See .Net's DataReaderParser.
+         */
+        String activityId = determineActivityId(httpResponse);
+        String message = errorFromResponse;
+        WebException formattedException = new WebException(errorFromResponse, httpResponse, thrownException);
+        boolean isPermanent = false;
+        if (!StringUtils.isBlank(errorFromResponse)) {
+            try {
+                JsonNode jsonObject = Utils.getObjectMapper().readTree(errorFromResponse);
+                if (jsonObject.has("error")) {
+                    formattedException = new DataWebException(errorFromResponse, httpResponse, thrownException);
+                    OneApiError apiError = ((DataWebException) formattedException).getApiError();
+                    message = apiError.getDescription();
+                    isPermanent = apiError.isPermanent();
+                } else if (jsonObject.has("message")) {
+                    message = jsonObject.get("message").asText();
+                }
+            } catch (JsonProcessingException e) {
+                // It's not ideal to use an exception here for control flow, but we can't know if it's a valid JSON until we try to parse it
+                LOGGER.debug("json processing error happened while parsing errorFromResponse {}", e.getMessage(), e);
+            }
+        } else {
+            message = String.format("Http StatusCode='%s'", httpResponse.getStatusCode());
+        }
+
+        return new DataServiceException(
+                url,
+                String.format("%s, ActivityId='%s'", message, activityId),
+                formattedException,
+                isPermanent);
+    }
+
+    private static void closeResourcesIfNeeded(boolean returnInputStream, HttpResponse httpResponse) {
+        // If we close the resources after returning the InputStream to the caller, they won't be able to read from it
+        if (!returnInputStream) {
+            if (httpResponse != null) {
+                httpResponse.close();
+            }
+        }
+    }
+
+    private static boolean shouldPostToOriginalUrlDueToRedirect(int redirectCount, int status) {
+        return (status == HttpStatus.FOUND || status == HttpStatus.TEMP_REDIRECT) && redirectCount + 1 <= MAX_REDIRECT_COUNT;
+    }
+
+    private static String determineActivityId(HttpResponse httpResponse) {
+        String activityId = "";
+
+        Optional<HttpHeader> activityIdHeader = Optional.ofNullable(
+                httpResponse.getHeaders().get(HttpHeaderName.fromString("x-ms-activity-id")));
+        if (activityIdHeader.isPresent()) {
+            activityId = activityIdHeader.get().getValue();
+        }
+        return activityId;
+    }
+
+}

data/src/main/java/com/microsoft/azure/kusto/data/ClientFactory.java

@@ -3,8 +3,9 @@
 
 package com.microsoft.azure.kusto.data;
 
+import com.azure.core.http.HttpClient;
 import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
-import org.apache.http.impl.client.CloseableHttpClient;
+import com.microsoft.azure.kusto.data.http.HttpClientProperties;
 
 import java.net.URISyntaxException;
 
@@ -43,12 +44,12 @@ public static Client createClient(ConnectionStringBuilder csb, HttpClientPropert
      * customized with the given properties.
      *
      * @param csb the connection string builder
-     * @param client CloseableHttpClient client. It will not be closed when {@link Client#close} is called.
+     * @param client HttpClient client.
      * @return a fully constructed {@linkplain Client} instance
      * @throws URISyntaxException if the cluster URL is invalid
      */
-    public static Client createClient(ConnectionStringBuilder csb, CloseableHttpClient client) throws URISyntaxException {
-        return client == null ? createClient(csb, (HttpClientProperties) null) : new ClientImpl(csb, client, true);
+    public static Client createClient(ConnectionStringBuilder csb, HttpClient client) throws URISyntaxException {
+        return client == null ? createClient(csb, (HttpClientProperties) null) : new ClientImpl(csb, client);
     }
 
     /**
@@ -85,7 +86,7 @@ public static StreamingClient createStreamingClient(ConnectionStringBuilder csb,
      * @return a fully constructed {@linkplain StreamingClient} instance
      * @throws URISyntaxException if the cluster URL is invalid
      */
-    public static StreamingClient createStreamingClient(ConnectionStringBuilder csb, CloseableHttpClient httpClient) throws URISyntaxException {
-        return new ClientImpl(csb, httpClient, true);
+    public static StreamingClient createStreamingClient(ConnectionStringBuilder csb, HttpClient httpClient) throws URISyntaxException {
+        return new ClientImpl(csb, httpClient);
     }
 }