Managed streaming - better estimation for decision for streaming vs queuing

CHANGELOG.md

@@ -4,7 +4,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [6.0.0] - 2024-09-21
+## [6.0.0] - 2024-10-27
+### Added
+- A new heuristic for choosing between queuing and streaming in Managed streaming client, the default behavior is using
+  an estimation against the 4mb limit after dividing or multiplying by some factor described by the consts:
+
+  This will also allow users to stream bigger than 4mb non-compressed data
+### Fixed
+- Some better error messages
+
 ### Changed
 - Replaced Apache CloseableHttpClient with configurable azure-core client.
 - [BREAKING] HttpClientFactory now accepts clients implementing azure-core HttpClient.

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

@@ -55,10 +55,6 @@ private String processResponseBody(HttpResponse response) throws DataServiceExce
         }
     }
 
-    protected InputStream postToStreamingOutput(HttpRequest request) throws DataServiceException {
-        return postToStreamingOutput(request, 0);
-    }
-
     // Todo: Implement async version of this method
     protected InputStream postToStreamingOutput(HttpRequest request, int redirectCount) throws DataServiceException {
 
@@ -143,7 +139,7 @@ public static DataServiceException createExceptionFromResponse(String url, HttpR
     }
 
     private static void closeResourcesIfNeeded(boolean returnInputStream, HttpResponse httpResponse) {
-        // If we close the resources after returning the InputStream to the user, he won't be able to read from it - used in streaming query 
+        // If we close the resources after returning the InputStream to the user, he won't be able to read from it - used in streaming query
         if (!returnInputStream) {
             if (httpResponse != null) {
                 httpResponse.close();

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

@@ -83,7 +83,7 @@ public static StreamingClient createStreamingClient(ConnectionStringBuilder csb,
      * is customized with the given properties.
      *
      * @param csb the connection string builder
-     * @param httpClient HTTP client
+     * @param httpClient HTTAP client
      * @return a fully constructed {@linkplain StreamingClient} instance
      * @throws URISyntaxException if the cluster URL is invalid
      */

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

@@ -11,12 +11,20 @@
 import com.microsoft.azure.kusto.data.auth.TokenProviderBase;
 import com.microsoft.azure.kusto.data.auth.TokenProviderFactory;
 import com.microsoft.azure.kusto.data.auth.endpoints.KustoTrustedEndpoints;
-import com.microsoft.azure.kusto.data.exceptions.*;
 import com.microsoft.azure.kusto.data.http.*;
 import com.microsoft.azure.kusto.data.instrumentation.*;
 import com.microsoft.azure.kusto.data.req.KustoRequest;
 import com.microsoft.azure.kusto.data.req.KustoRequestContext;
 import com.microsoft.azure.kusto.data.res.JsonResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import com.microsoft.azure.kusto.data.exceptions.KustoServiceQueryError;
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
+import com.microsoft.azure.kusto.data.http.HttpClientFactory;
+import com.microsoft.azure.kusto.data.http.UncloseableStream;
+import com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity;
+import com.microsoft.azure.kusto.data.instrumentation.SupplierTwoExceptions;
+import com.microsoft.azure.kusto.data.instrumentation.TraceableAttributes;
 import org.apache.commons.lang3.StringUtils;
 
 import org.jetbrains.annotations.NotNull;
@@ -119,9 +127,9 @@ private KustoOperationResult processJsonResult(JsonResult res) throws DataServic
             return new KustoOperationResult(res.getResult(), res.getEndpoint().endsWith("v2/rest/query") ? "v2" : "v1");
         } catch (KustoServiceQueryError e) {
             throw new DataServiceException(res.getEndpoint(),
-                    "Error found while parsing json response as KustoOperationResult:" + e.getMessage(), e, e.isPermanent());
+                    "Error found while parsing json response as KustoOperationResult:" + e, e, e.isPermanent());
         } catch (Exception e) {
-            throw new DataClientException(res.getEndpoint(), e.getMessage(), e);
+            throw new DataClientException(res.getEndpoint(), ExceptionsUtils.getMessageEx(e), e);
         }
     }
 
@@ -367,5 +375,4 @@ public String getClusterUrl() {
     ClientDetails getClientDetails() {
         return clientDetails;
     }
-
 }

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

@@ -40,7 +40,7 @@
 public class ClientRequestProperties implements Serializable, TraceableAttributes {
     public static final String OPTION_SERVER_TIMEOUT = "servertimeout";
 
-    //  If set and positive, indicates the maximum number of HTTP redirects that the client will process. [Integer]
+    // If set and positive, indicates the maximum number of HTTP redirects that the client will process. [Integer]
     public static final String OPTION_CLIENT_MAX_REDIRECT_COUNT = "client_max_redirect_count";
     /*
      * Matches valid Kusto Timespans: Optionally negative, optional number of days followed by a period, optionally up to 24 as hours followed by a colon,

data/src/main/java/com/microsoft/azure/kusto/data/auth/CallbackTokenProvider.java

@@ -6,6 +6,7 @@
 import com.azure.core.http.HttpClient;
 import com.microsoft.azure.kusto.data.exceptions.DataClientException;
 
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -32,7 +33,7 @@ protected String acquireAccessTokenImpl() throws DataClientException {
         try {
             return tokenProvider.apply(httpClient);
         } catch (Exception e) {
-            throw new DataClientException(clusterUrl, e.getMessage(), e);
+            throw new DataClientException(clusterUrl, ExceptionsUtils.getMessageEx(e), e);
         }
     }
 

data/src/main/java/com/microsoft/azure/kusto/data/auth/CloudInfo.java

@@ -8,6 +8,7 @@
 import com.microsoft.azure.kusto.data.ExponentialRetry;
 import com.microsoft.azure.kusto.data.Utils;
 import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import com.microsoft.azure.kusto.data.http.HttpClientFactory;
 import com.microsoft.azure.kusto.data.UriUtils;
 import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
@@ -100,7 +101,8 @@ public static CloudInfo retrieveCloudInfoForCluster(String clusterUrl,
                     throw new DataServiceException(clusterUrl, "URISyntaxException when trying to retrieve cluster metadata:" + e.getMessage(), e, true);
                 } catch (IOException ex) {
                     if (!Utils.isRetriableIOException(ex)) {
-                        throw new DataServiceException(clusterUrl, "IOException when trying to retrieve cluster metadata:" + ex.getMessage(), ex,
+                        throw new DataServiceException(clusterUrl, "IOException when trying to retrieve cluster metadata:" + ExceptionsUtils.getMessageEx(ex),
+                                ex,
                                 Utils.isRetriableIOException(ex));
                     }
                 } catch (DataServiceException e) {

data/src/main/java/com/microsoft/azure/kusto/data/auth/endpoints/WellKnownKustoEndpointsData.java

@@ -44,8 +44,7 @@ private static WellKnownKustoEndpointsData readInstance() {
                 return objectMapper.readValue(resourceAsStream, WellKnownKustoEndpointsData.class);
             }
         } catch (Exception ex) {
-            ex.printStackTrace();
+            throw new RuntimeException("Failed to read WellKnownKustoEndpoints.json", ex);
         }
-        return null;
     }
 }

data/src/main/java/com/microsoft/azure/kusto/data/exceptions/DataWebException.java

@@ -35,9 +35,10 @@ public OneApiError getApiError() {
             try {
                 apiError = OneApiError.fromJsonObject(objectMapper.readTree(getMessage()).get("error"));
             } catch (JsonProcessingException e) {
-                log.error("failed to parse error from message {} {} ", e.getMessage(), e);
+                log.error(String.format("failed to parse error from message: '%s' ", e.getMessage()), e);
             }
         }
+
         return apiError;
     }
 }

data/src/main/java/com/microsoft/azure/kusto/data/exceptions/ExceptionsUtils.java

@@ -0,0 +1,8 @@
+package com.microsoft.azure.kusto.data.exceptions;
+
+public class ExceptionsUtils {
+    // Useful in IOException, where message might not propagate to the base IOException
+    public static String getMessageEx(Exception e) {
+        return (e.getMessage() == null && e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+    }
+}

data/src/main/java/com/microsoft/azure/kusto/data/http/HttpClientFactory.java

@@ -67,5 +67,4 @@ public static HttpClient create(HttpClientProperties properties) {
 
         return HttpClient.createDefault(options);
     }
-
 }

data/src/main/java/com/microsoft/azure/kusto/data/instrumentation/MonitoredActivity.java

@@ -2,7 +2,6 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.Callable;
 
 public class MonitoredActivity {
     public static void invoke(Runnable runnable, String nameOfSpan) {

ingest/pom.xml

@@ -108,6 +108,7 @@
                         <dependency>org.reactivestreams:reactive-streams:jar</dependency>
                         <dependency>com.microsoft.azure:msal4j:jar</dependency>
                         <dependency>io.projectreactor:reactor-core:jar</dependency>
+                        <dependency>com.fasterxml.jackson.core:jackson-core:jar</dependency>
                     </ignoredUsedUndeclaredDependencies>
                     <ignoreNonCompile>true</ignoreNonCompile>
                     <ignoredNonTestScopedDependencies>

ingest/src/main/java/com/microsoft/azure/kusto/ingest/IngestClientBase.java

@@ -1,11 +1,11 @@
 package com.microsoft.azure.kusto.ingest;
 
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import com.microsoft.azure.kusto.ingest.source.CompressionType;
 import org.apache.http.conn.util.InetAddressUtils;
 
-import java.net.InetAddress;
+import java.io.IOException;
 import java.net.URI;
-import java.net.UnknownHostException;
 import com.microsoft.azure.kusto.data.instrumentation.SupplierTwoExceptions;
 import com.microsoft.azure.kusto.data.instrumentation.TraceableAttributes;
 import com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity;
@@ -130,14 +130,20 @@ public IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceIn
      * @see IngestionProperties
      */
     protected abstract IngestionResult ingestFromStreamImpl(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties)
-            throws IngestionClientException, IngestionServiceException;
+            throws IngestionClientException, IngestionServiceException, IOException;
 
     public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties)
             throws IngestionClientException, IngestionServiceException {
         // trace ingestFromStream
         return MonitoredActivity.invoke(
-                (SupplierTwoExceptions<IngestionResult, IngestionClientException, IngestionServiceException>) () -> ingestFromStreamImpl(streamSourceInfo,
-                        ingestionProperties),
+                (SupplierTwoExceptions<IngestionResult, IngestionClientException, IngestionServiceException>) () -> {
+                    try {
+                        return ingestFromStreamImpl(streamSourceInfo,
+                                ingestionProperties);
+                    } catch (IOException e) {
+                        throw new IngestionServiceException(ExceptionsUtils.getMessageEx(e), e);
+                    }
+                },
                 getClientType().concat(".ingestFromStream"));
     }
 

ingest/src/main/java/com/microsoft/azure/kusto/ingest/IngestionProperties.java

@@ -431,6 +431,10 @@ public IngestionMapping.IngestionMappingKind getIngestionMappingKind() {
         public boolean isCompressible() {
             return compressible;
         }
+
+        public boolean isJsonFormat() {
+            return this.equals(JSON) || this.equals(MULTIJSON) || this.equals(SINGLEJSON);
+        }
     }
 
     public enum IngestionReportLevel {