Paginated Directory Delete

sdk/storage/Azure.Storage.Files.DataLake/src/DataLakePathClient.cs

@@ -1860,31 +1860,47 @@ private async Task<Response> DeleteInternal(
                 try
                 {
                     scope.Start();
-                    ResponseWithHeaders<PathDeleteHeaders> response;
+                    ResponseWithHeaders<PathDeleteHeaders> response = null;
 
-                    if (async)
+                    // Pagination only applies to service version 2023-08-03 and later, when using OAuth.
+                    bool? paginated = null;
+                    if (_clientConfiguration.ClientOptions.Version >= DataLakeClientOptions.ServiceVersion.V2023_08_03
+                        && _clientConfiguration.TokenCredential != null)
                     {
-                        response = await PathRestClient.DeleteAsync(
-                            recursive: recursive,
-                            leaseId: conditions?.LeaseId,
-                            ifMatch: conditions?.IfMatch?.ToString(),
-                            ifNoneMatch: conditions?.IfNoneMatch?.ToString(),
-                            ifModifiedSince: conditions?.IfModifiedSince,
-                            ifUnmodifiedSince: conditions?.IfUnmodifiedSince,
-                            cancellationToken: cancellationToken)
-                            .ConfigureAwait(false);
+                        paginated = true;
                     }
-                    else
+
+                    do
                     {
-                        response = PathRestClient.Delete(
-                            recursive: recursive,
-                            leaseId: conditions?.LeaseId,
-                            ifMatch: conditions?.IfMatch?.ToString(),
-                            ifNoneMatch: conditions?.IfNoneMatch?.ToString(),
-                            ifModifiedSince: conditions?.IfModifiedSince,
-                            ifUnmodifiedSince: conditions?.IfUnmodifiedSince,
-                            cancellationToken: cancellationToken);
+                        if (async)
+                        {
+                            response = await PathRestClient.DeleteAsync(
+                                recursive: recursive,
+                                continuation: response?.Headers?.Continuation,
+                                leaseId: conditions?.LeaseId,
+                                ifMatch: conditions?.IfMatch?.ToString(),
+                                ifNoneMatch: conditions?.IfNoneMatch?.ToString(),
+                                ifModifiedSince: conditions?.IfModifiedSince,
+                                ifUnmodifiedSince: conditions?.IfUnmodifiedSince,
+                                paginated: paginated,
+                                cancellationToken: cancellationToken)
+                                .ConfigureAwait(false);
+                        }
+                        else
+                        {
+                            response = PathRestClient.Delete(
+                                recursive: recursive,
+                                continuation: response?.Headers?.Continuation,
+                                leaseId: conditions?.LeaseId,
+                                ifMatch: conditions?.IfMatch?.ToString(),
+                                ifNoneMatch: conditions?.IfNoneMatch?.ToString(),
+                                ifModifiedSince: conditions?.IfModifiedSince,
+                                ifUnmodifiedSince: conditions?.IfUnmodifiedSince,
+                                paginated: paginated,
+                                cancellationToken: cancellationToken);
+                        }
                     }
+                    while (!string.IsNullOrEmpty(response?.Headers?.Continuation));
 
                     return response.GetRawResponse();
                 }
@@ -3177,7 +3193,7 @@ private async Task<Response<AccessControlChangeResult>> SetAccessControlRecursiv
                             }
                             batchesCount++;
                         } while (!string.IsNullOrEmpty(continuationToken)
-                            && (!options.MaxBatches.HasValue || batchesCount < options.MaxBatches.Value));
+                            && (options == null || !options.MaxBatches.HasValue || batchesCount < options.MaxBatches.Value));
 
                         return Response.FromValue(
                             new AccessControlChangeResult()

sdk/storage/Azure.Storage.Files.DataLake/src/autorest.md

@@ -4,7 +4,7 @@ Run `dotnet build /t:GenerateCode` to generate code.
 
 ``` yaml
 input-file:
-    - https://raw.githubusercontent.com/Azure/azure-rest-api-specs/3e6f238a1f74d77ba6970f297c77995a9f1f374e/specification/storage/data-plane/Azure.Storage.Files.DataLake/preview/2021-06-08/DataLakeStorage.json
+    - https://raw.githubusercontent.com/Azure/azure-rest-api-specs/2918e938227a026731c9ac4788c24c57e30b257b/specification/storage/data-plane/Azure.Storage.Files.DataLake/preview/2023-05-03/DataLakeStorage.json
 generation1-convenience-client: true
 modelerfour:
     seal-single-value-enum-by-default: true

sdk/storage/Azure.Storage.Files.DataLake/tests/DirectoryClientTests.cs

@@ -1019,6 +1019,52 @@ await TestHelper.AssertExpectedExceptionAsync<RequestFailedException>(
             }
         }
 
+        // To run this test, the NamespaceTenant AAD info needs to be set to an AAD app that does not have any RBAC permissions,
+        // and entityId needs to be set to the entity ID of the application.
+        [RecordedTest]
+        [Ignore("AAD app not configured for this test")]
+        [ServiceVersion(Min = DataLakeClientOptions.ServiceVersion.V2023_08_03)]
+        public async Task DeleteAsync_Paginated()
+        {
+            // object ID of AAD app that has no RBAC permissions
+            string entityId = "a251dfc9-65c7-4ec3-84d8-f3f3bec0a96f";
+            string fileSystemName = GetNewFileSystemName();
+            await using DisposingFileSystem test = await GetNewFileSystem(fileSystemName: fileSystemName);
+
+            // Arrange
+            string directoryName = GetNewDirectoryName();
+            DataLakeDirectoryClient directory = InstrumentClient(test.FileSystem.GetDirectoryClient(directoryName));
+            await directory.CreateIfNotExistsAsync();
+
+            for (int i = 0; i < 5020; i++)
+            {
+                DataLakeFileClient fileClient = directory.GetFileClient(GetNewFileName());
+                await fileClient.CreateIfNotExistsAsync();
+            };
+
+            DataLakeDirectoryClient rootDirectory = InstrumentClient(test.FileSystem.GetDirectoryClient("/"));
+
+            Response<PathAccessControl> aclResponse = await rootDirectory.GetAccessControlAsync();
+
+            IList<PathAccessControlItem> accessControlList = aclResponse.Value.AccessControlList.ToList();
+            accessControlList.Add(
+                new PathAccessControlItem
+                {
+                    Permissions = RolePermissions.Read | RolePermissions.Write | RolePermissions.Execute,
+                    AccessControlType = AccessControlType.User,
+                    EntityId = entityId
+                });
+
+            await rootDirectory.SetAccessControlRecursiveAsync(accessControlList);
+
+            DataLakeServiceClient oauthService = GetServiceClient_OAuth();
+            DataLakeFileSystemClient oauthFileSystem = InstrumentClient(oauthService.GetFileSystemClient(fileSystemName));
+            DataLakeDirectoryClient oauthDirectory = InstrumentClient(oauthFileSystem.GetDirectoryClient(directoryName));
+
+            // Act
+            Response response = await oauthDirectory.DeleteAsync();
+        }
+
         [RecordedTest]
         public async Task RenameAsync()
         {