chore: Merge release/4.6.1 to master (#138)

Author: hopengfoong

docs/4.6.1-release-notes.md

@@ -0,0 +1,3 @@
+### Features
+- Additional information output to help diagnose common azure issues
+- Handle change in microsoft's libraries that resulted in a 404 when writing to a new file

src/Connector.DataLake.Common/Connector/DataLakeExportEntitiesJobBase.cs

@@ -10,6 +10,7 @@
 using Azure.Storage.Files.DataLake;
 
 using CluedIn.Connector.DataLake.Common.Connector.SqlDataWriter;
+using CluedIn.Connector.DataLake.Common.Extensions;
 using CluedIn.Core;
 using CluedIn.Core.Data.Relational;
 using CluedIn.Core.Streams;
@@ -164,12 +165,26 @@ public override async Task DoRunAsync(ExecutionContext context, IDataLakeJobArgs
             [DataTimeKey] = asOfTime,
         });
 
+        DataLakeFileClient temporaryFileClient;
+        try
+        {
+            temporaryFileClient = directoryClient.GetFileClient(temporaryOutputFileName);
+        }
+        catch
+        {
+            context.Log.LogInformation(
+                "Error creating file client for {TemporaryOutputFileName}.",
+                temporaryOutputFileName);
+            throw;
+        }
+
         context.Log.LogInformation(
-            "Begin writing to file '{OutputFileName}' using data at {DataTime} and {TemporaryOutputFileName}.",
+            "Begin writing to file '{OutputFileName}' using data at {DataTime} and {TemporaryOutputFileName} ({TemporaryFileClientUri}).",
             outputFileName,
             asOfTime,
-            temporaryOutputFileName);
-        var temporaryFileClient = directoryClient.GetFileClient(temporaryOutputFileName);
+            temporaryOutputFileName,
+            temporaryFileClient.Uri);
+
         var totalRows = await writeFileContentsAsync();
         if (configuration.IsDeltaMode && totalRows == 0 && !GetIsEmptyFileAllowed(exportJobData))
         {
@@ -213,7 +228,7 @@ async Task<long> writeFileContentsAsync()
         {
             var fieldNamesToUse = await GetFieldNamesAsync(context, exportJobData, configuration, fieldNames);
             var sqlDataWriter = GetSqlDataWriter(outputFormat);
-            await using var outputStream = await temporaryFileClient.OpenWriteAsync(configuration.IsOverwriteEnabled);
+            await using var outputStream = await temporaryFileClient.OpenWriteExAsync(configuration.IsOverwriteEnabled);
             using var bufferedStream = new DataLakeBufferedWriteStream(outputStream);
             return await sqlDataWriter?.WriteAsync(context, configuration, bufferedStream, fieldNamesToUse, IsInitialExport, reader);
         }

src/Connector.DataLake.Common/Extensions/DataLakeFileClientExtensions.cs

@@ -0,0 +1,45 @@
+using Azure.Storage.Files.DataLake;
+using Azure;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Storage.Files.DataLake.Models;
+
+namespace CluedIn.Connector.DataLake.Common.Extensions;
+
+public static class DataLakeFileClientExtensions
+{
+    /// <summary>
+    /// Opens a stream for writing to the Data Lake file, safely handling the case where 
+    /// OpenWriteAsync fails with a 404 because the file does not yet exist.
+    /// This pattern is necessary due to the DataLake SDK's internal GetProperties check.
+    /// </summary>
+    /// <param name="fileClient">The DataLakeFileClient instance.</param>
+    /// <param name="overwrite">Whether to overwrite the file if it exists.</param>
+    /// <param name="options">Optional parameters for the write operation.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>A writable Stream for the file.</returns>
+    public static async Task<Stream> OpenWriteExAsync(
+        this DataLakeFileClient fileClient,
+        bool overwrite,
+        DataLakeFileOpenWriteOptions options = null,
+        CancellationToken cancellationToken = default)
+    {
+        try
+        {
+            // 1. Attempt the standard OpenWriteAsync call.
+            return await fileClient.OpenWriteAsync(overwrite, options, cancellationToken);
+        }
+        catch (RequestFailedException ex) when (ex.Status == 404)
+        {
+            // This is the specific error when GetPropertiesAsync fails because the file doesn't exist.
+
+            await fileClient.CreateAsync(cancellationToken: cancellationToken);
+
+            // Retry the OpenWriteAsync call. This time, GetPropertiesAsync should succeed 
+            // (because the file exists now), and the stream will be opened.
+            // We pass the original 'overwrite' flag here to handle the stream logic correctly.
+            return await fileClient.OpenWriteAsync(overwrite, options, cancellationToken);
+        }
+    }
+}

test/integration/Connector.DataLake.Common.Tests.Integration/DataLakeConnectorTestsBase.cs

@@ -218,6 +218,7 @@ protected virtual StreamModel CreateStreamModel(Guid providerDefinitionId, Strea
             ExportIncomingEdges = true,
             ExportOutgoingEdges = true,
             Status = StreamStatus.Started,
+            OrganizationId = Guid.NewGuid(),    // can't be Guid.Empty
         };
         return streamModel;
     }