Limiting batch size for table storage

src/Akka.Persistence.Azure/CloudTableExtensions.cs

@@ -0,0 +1,46 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.WindowsAzure.Storage.Table;
+
+namespace Akka.Persistence.Azure
+{
+    public static class CloudTableExtensions
+    {
+        private const int MaxBatchSize = 100;
+
+        public static async Task<IList<TableResult>> ExecuteBatchAsLimitedBatches(
+            this CloudTable table,
+            TableBatchOperation batch)
+        {
+            if (batch.Count < 1)
+                return new List<TableResult>();
+
+            if (batch.Count <= MaxBatchSize)
+                return await table.ExecuteBatchAsync(batch);
+
+            var result = new List<TableResult>();
+            var limitedBatchOperationLists = batch.ChunkBy(MaxBatchSize);
+
+            foreach (var limitedBatchOperationList in limitedBatchOperationLists)
+            {
+                var limitedBatch = CreateLimitedTableBatchOperation(limitedBatchOperationList);
+                var limitedBatchResult = await table.ExecuteBatchAsync(limitedBatch);
+                result.AddRange(limitedBatchResult);
+            }
+
+            return result;
+        }
+
+        private static TableBatchOperation CreateLimitedTableBatchOperation(
+            IEnumerable<TableOperation> limitedBatchOperationList)
+        {
+            var limitedBatch = new TableBatchOperation();
+            foreach (var limitedBatchOperation in limitedBatchOperationList)
+            {
+                limitedBatch.Add(limitedBatchOperation);
+            }
+
+            return limitedBatch;
+        }
+    }
+}

src/Akka.Persistence.Azure/Journal/AzureTableStorageJournal.cs

@@ -223,7 +223,7 @@ protected override async Task DeleteMessagesToAsync(
                         tableBatchOperation.Delete(toBeDeleted);
                     }
 
-                    var deleteTask = Table.ExecuteBatchAsync(tableBatchOperation);
+                    var deleteTask = Table.ExecuteBatchAsLimitedBatches(tableBatchOperation);
 
                     await deleteTask;
                 }
@@ -378,7 +378,7 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(
                             if (_log.IsDebugEnabled && _settings.VerboseLogging)
                                 _log.Debug("Attempting to write batch of {0} messages to Azure storage", persistenceBatch.Count);
 
-                            var persistenceResults = await Table.ExecuteBatchAsync(persistenceBatch);
+                            var persistenceResults = await Table.ExecuteBatchAsLimitedBatches(persistenceBatch);
 
                             if (_log.IsDebugEnabled && _settings.VerboseLogging)
                                 foreach (var r in persistenceResults)
@@ -405,7 +405,7 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(
                         allPersistenceIdsBatch.InsertOrReplace(new AllPersistenceIdsEntry(encodedKey));
                     });
 
-                    var allPersistenceResults = await Table.ExecuteBatchAsync(allPersistenceIdsBatch);
+                    var allPersistenceResults = await Table.ExecuteBatchAsLimitedBatches(allPersistenceIdsBatch);
 
                     if (_log.IsDebugEnabled && _settings.VerboseLogging)
                         foreach (var r in allPersistenceResults)
@@ -430,7 +430,7 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(
                                 eventTagsBatch.InsertOrReplace(item);
                             }
 
-                            var eventTagsResults = await Table.ExecuteBatchAsync(eventTagsBatch);
+                            var eventTagsResults = await Table.ExecuteBatchAsLimitedBatches(eventTagsBatch);
 
                             if (_log.IsDebugEnabled && _settings.VerboseLogging)
                                 foreach (var r in eventTagsResults)

src/Akka.Persistence.Azure/ListExtensions.cs

@@ -0,0 +1,18 @@
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Linq;
+
+namespace Akka.Persistence.Azure
+{
+    public static class ListExtensions
+    {
+        public static IImmutableList<IEnumerable<T>> ChunkBy<T>(this IEnumerable<T> source, int chunkSize)
+        {
+            return source
+                .Select((x, i) => new { Index = i, Value = x })
+                .GroupBy(x => x.Index / chunkSize)
+                .Select(x => x.Select(v => v.Value).ToList().AsEnumerable())
+                .ToImmutableList();
+        }
+    }
+}