new clientRequestID per call

Azure · Jan 28, 2025 · cebc2b5 · cebc2b5
1 parent 6684d15
commit cebc2b5
Show file tree

Hide file tree

Showing 7 changed files with 96 additions and 70 deletions.
diff --git a/connector/scalastyle_config.xml b/connector/scalastyle_config.xml
@@ -86,7 +86,7 @@
  </check>
  <check level="warning" class="org.scalastyle.scalariform.MethodLengthChecker" enabled="true">
   <parameters>
-   <parameter name="maxLength"><![CDATA[70]]></parameter>
+   <parameter name="maxLength"><![CDATA[120]]></parameter>
   </parameters>
  </check>
  <check level="warning" class="org.scalastyle.scalariform.MethodNamesChecker" enabled="true">

diff --git a/connector/src/main/scala/com/microsoft/kusto/spark/datasink/FinalizeHelper.scala b/connector/src/main/scala/com/microsoft/kusto/spark/datasink/FinalizeHelper.scala
@@ -103,6 +103,7 @@ object FinalizeHelper {
                   tmpTableName,
                   allowMerge = false,
                   allowRebuild = false),
+                "alterMergePolicyCommand",
                 crp)
               // Drop dedup tags
               if (writeOptions.ensureNoDupBlobs) {
@@ -112,6 +113,7 @@ object FinalizeHelper {
                   crp,
                   writeOptions.timeout,
                   s"drops extents from temp table '$tmpTableName' ",
+                  "extentsDrop",
                   writeOptions.requestId)
               }
               client.moveExtents(

diff --git a/connector/src/main/scala/com/microsoft/kusto/spark/datasink/KustoWriter.scala b/connector/src/main/scala/com/microsoft/kusto/spark/datasink/KustoWriter.scala
@@ -101,6 +101,7 @@ object KustoWriter {
       .executeEngine(
         tableCoordinates.database,
         generateTableGetSchemaAsRowsCommand(tableCoordinates.table.get),
+        "schemaShow",
         crp)
       .getPrimaryResults
 
@@ -131,6 +132,7 @@ object KustoWriter {
             .executeEngine(
               tableCoordinates.database,
               generateTableGetSchemaAsRowsCommand(writeOptions.userTempTableName.get),
+              "schemaShow",
               crp)
             .getPrimaryResults
             .count() <= 0 ||
@@ -449,7 +451,7 @@ object KustoWriter {
       streamingClient: ManagedStreamingIngestClient,
       inputStreamLastIdx: Int): Unit = {
     KDSU.retryApplyFunction(
-      () => {
+      i => {
         val inputStream = new ByteArrayInputStream(bytes, 0, inputStreamLastIdx)
         // The SDK will compress the stream by default.
         val streamSourceInfo = new StreamSourceInfo(inputStream)
@@ -463,7 +465,8 @@ object KustoWriter {
                   s"details: ${ingestionStatus.details}, " +
                   s"activityId: ${ingestionStatus.activityId}, " +
                   s"errorCode: ${ingestionStatus.errorCode}, " +
-                  s"errorCodeString: ${ingestionStatus.errorCodeString}" +
+                  s"errorCodeString: ${ingestionStatus.errorCodeString}," +
+                  s"retry: $i" +
                   "}")
             })
           case Failure(e: Throwable) =>
@@ -583,7 +586,7 @@ object KustoWriter {
       }
       // write the data here
       val partitionsResult = KDSU.retryApplyFunction(
-        () => {
+        i => {
           Try(
             ingestClient.ingestFromBlob(
               new BlobSourceInfo(blobUri + sas, size, UUID.randomUUID()),
@@ -599,7 +602,7 @@ object KustoWriter {
               KDSU.reportExceptionAndThrow(
                 className,
                 e,
-                "Queueing blob for ingestion in partition " +
+                s"Queueing blob for ingestion, retry number '$i', in partition " +
                   s"$partitionIdString for requestId: '${parameters.writeOptions.requestId}")
               val blobUrlWithSas =
                 s"${blobResource.blob.getStorageUri.getPrimaryUri.toString}${blobResource.sas}"

diff --git a/connector/src/main/scala/com/microsoft/kusto/spark/datasource/KustoReader.scala b/connector/src/main/scala/com/microsoft/kusto/spark/datasource/KustoReader.scala
@@ -84,6 +84,7 @@ private[kusto] object KustoReader {
       .executeEngine(
         request.kustoCoordinates.database,
         filteredQuery,
+        "executeQuery",
         request.clientRequestProperties.orNull)
       .getPrimaryResults
 
@@ -329,6 +330,7 @@ private[kusto] class KustoReader(client: ExtendedKustoClient) {
       .executeEngine(
         request.kustoCoordinates.database,
         exportCommand,
+        "exportPartitionToBlob",
         request.clientRequestProperties.orNull)
       .getPrimaryResults
     KDSU.verifyAsyncCommandCompletion(

diff --git a/connector/src/main/scala/com/microsoft/kusto/spark/utils/ContainerProvider.scala b/connector/src/main/scala/com/microsoft/kusto/spark/utils/ContainerProvider.scala
@@ -83,7 +83,7 @@ class ContainerProvider(
 
   private def refresh(exportContainer: Boolean = false): ContainerAndSas = {
     if (exportContainer) {
-      Try(client.executeDM(command, None, Some(retryConfigExportContainers))) match {
+      Try(client.executeDM(command, None, "refreshContainers", Some(retryConfigExportContainers))) match {
         case Success(res) =>
           val storage = res.getPrimaryResults.getData.asScala.map(row => {
             val parts = row.get(0).toString.split('?')