Skip to content

Commit ac47852

Browse files
ArkatufusArkatufus
authored
Re-add chunked transaction batching support (#254)
* Re-add chunked transaction batching support * Add unit tests * Fix unit test
1 parent d1baecf commit ac47852

File tree

3 files changed

+112
-12
lines changed

3 files changed

+112
-12
lines changed

‎src/Akka.Persistence.Azure.Tests/LimitedBatchSpec.cs‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
// -----------------------------------------------------------------------
2+
// <copyright file="LimitedBatchSpec.cs" company="Akka.NET Project">
3+
// Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
4+
// Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
5+
// </copyright>
6+
// -----------------------------------------------------------------------
7+
8+
using System.Collections.Generic;
9+
using System.Linq;
10+
using System.Threading;
11+
using System.Threading.Tasks;
12+
using Akka.Persistence.Azure.Tests.Helper;
13+
using Azure.Data.Tables;
14+
using FluentAssertions;
15+
using FluentAssertions.Extensions;
16+
using Xunit;
17+
using Xunit.Abstractions;
18+
19+
namespace Akka.Persistence.Azure.Tests
20+
{
21+
public class LimitedBatchSpec: IAsyncLifetime
22+
{
23+
private readonly TableClient _tableClient;
24+
25+
public LimitedBatchSpec(ITestOutputHelper output)
26+
{
27+
_tableClient = new TableClient("UseDevelopmentStorage=true", "testtable");
28+
}
29+
30+
public async Task InitializeAsync()
31+
{
32+
await DbUtils.CleanupCloudTable("UseDevelopmentStorage=true");
33+
await _tableClient.CreateAsync();
34+
}
35+
36+
public Task DisposeAsync()
37+
{
38+
return Task.CompletedTask;
39+
}
40+
41+
[Fact(DisplayName = "Limited batch with 0 entries should return empty list")]
42+
public async Task ZeroEntriesTest()
43+
{
44+
using var cts = new CancellationTokenSource(3.Seconds());
45+
var result = await _tableClient.ExecuteBatchAsLimitedBatches(new List<TableTransactionAction>(), cts.Token);
46+
result.Count.Should().Be(0);
47+
48+
var entities = await _tableClient.QueryAsync<TableEntity>("PartitionKey eq 'test'", null, null, cts.Token)
49+
.ToListAsync(cts.Token);
50+
entities.Count.Should().Be(0);
51+
}
52+
53+
[Fact(DisplayName = "Limited batch with less than 100 entries should work")]
54+
public async Task FewEntriesTest()
55+
{
56+
var entries = Enumerable.Range(1, 50)
57+
.Select(i => new TableTransactionAction(TableTransactionActionType.Add, new TableEntity
58+
{
59+
PartitionKey = "test",
60+
RowKey = i.ToString("D8")
61+
})).ToList();
62+
63+
using var cts = new CancellationTokenSource(3.Seconds());
64+
var result = await _tableClient.ExecuteBatchAsLimitedBatches(entries, cts.Token);
65+
result.Count.Should().Be(50);
66+
67+
var entities = await _tableClient.QueryAsync<TableEntity>("PartitionKey eq 'test'", null, null, cts.Token)
68+
.ToListAsync(cts.Token);
69+
entities.Count.Should().Be(50);
70+
entities.Select(e => int.Parse(e.RowKey.TrimStart('0'))).Should().BeEquivalentTo(Enumerable.Range(1, 50));
71+
}
72+
73+
[Fact(DisplayName = "Limited batch with more than 100 entries should work")]
74+
public async Task LotsEntriesTest()
75+
{
76+
var entries = Enumerable.Range(1, 505)
77+
.Select(i => new TableTransactionAction(TableTransactionActionType.Add, new TableEntity
78+
{
79+
PartitionKey = "test",
80+
RowKey = i.ToString("D8")
81+
})).ToList();
82+
83+
using var cts = new CancellationTokenSource(3.Seconds());
84+
var result = await _tableClient.ExecuteBatchAsLimitedBatches(entries, cts.Token);
85+
result.Count.Should().Be(505);
86+
87+
var entities = await _tableClient.QueryAsync<TableEntity>("PartitionKey eq 'test'", null, null, cts.Token)
88+
.ToListAsync(cts.Token);
89+
entities.Count.Should().Be(505);
90+
entities.Select(e => int.Parse(e.RowKey.TrimStart('0'))).Should().BeEquivalentTo(Enumerable.Range(1, 505));
91+
}
92+
}
93+
}

‎src/Akka.Persistence.Azure/CloudTableExtensions.cs‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
using System.Collections.Generic;
88
using System.Collections.Immutable;
9+
using System.Threading;
910
using System.Threading.Tasks;
1011
using Azure;
1112
using Azure.Data.Tables;
@@ -18,20 +19,21 @@ public static class CloudTableExtensions
1819

1920
public static async Task<IReadOnlyList<Response>> ExecuteBatchAsLimitedBatches(
2021
this TableClient table,
21-
List<TableTransactionAction> batch)
22+
List<TableTransactionAction> batch,
23+
CancellationToken token)
2224
{
2325
if (batch.Count < 1)
2426
return ImmutableList<Response>.Empty;
2527

2628
if (batch.Count <= MaxBatchSize)
27-
return (await table.SubmitTransactionAsync(batch)).Value;
29+
return (await table.SubmitTransactionAsync(batch, token)).Value;
2830

2931
var result = new List<Response>();
3032
var limitedBatchOperationLists = batch.ChunkBy(MaxBatchSize);
3133

3234
foreach (var limitedBatchOperationList in limitedBatchOperationLists)
3335
{
34-
var limitedBatchResponse = await table.SubmitTransactionAsync(limitedBatchOperationList);
36+
var limitedBatchResponse = await table.SubmitTransactionAsync(limitedBatchOperationList, token);
3537
result.AddRange(limitedBatchResponse.Value);
3638
}
3739

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

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -229,10 +229,15 @@ protected override async Task DeleteMessagesToAsync(string persistenceId, long t
229229
else
230230
nextTask = null;
231231

232-
if (currentPage.Values.Count > 0)
232+
var response = await Table.ExecuteBatchAsLimitedBatches(currentPage.Values
233+
.Select(entity => new TableTransactionAction(TableTransactionActionType.Delete, entity)).ToList(), _shutdownCts.Token);
234+
235+
if (_log.IsDebugEnabled && _settings.VerboseLogging)
233236
{
234-
await Table.SubmitTransactionAsync(currentPage.Values
235-
.Select(entity => new TableTransactionAction(TableTransactionActionType.Delete, entity)), _shutdownCts.Token);
237+
foreach (var r in response)
238+
{
239+
_log.Debug("Azure table storage wrote entities with status code [{0}]", r.Status);
240+
}
236241
}
237242
}
238243

@@ -365,10 +370,10 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(IEnu
365370
if (_log.IsDebugEnabled && _settings.VerboseLogging)
366371
_log.Debug("Attempting to write batch of {0} messages to Azure storage", batchItems.Count);
367372

368-
var response = await Table.SubmitTransactionAsync(batchItems, _shutdownCts.Token);
373+
var response = await Table.ExecuteBatchAsLimitedBatches(batchItems, _shutdownCts.Token);
369374
if (_log.IsDebugEnabled && _settings.VerboseLogging)
370375
{
371-
foreach (var r in response.Value)
376+
foreach (var r in response)
372377
{
373378
_log.Debug("Azure table storage wrote entities with status code [{0}]", r.Status);
374379
}
@@ -395,10 +400,10 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(IEnu
395400
new AllPersistenceIdsEntry(PartitionKeyEscapeHelper.Escape(item.Key)).WriteEntity()));
396401
}
397402

398-
var allPersistenceResponse = await Table.SubmitTransactionAsync(allPersistenceIdsBatch, _shutdownCts.Token);
403+
var allPersistenceResponse = await Table.ExecuteBatchAsLimitedBatches(allPersistenceIdsBatch, _shutdownCts.Token);
399404

400405
if (_log.IsDebugEnabled && _settings.VerboseLogging)
401-
foreach (var r in allPersistenceResponse.Value)
406+
foreach (var r in allPersistenceResponse)
402407
_log.Debug("Azure table storage wrote entity with status code [{0}]", r.Status);
403408

404409
if (HasPersistenceIdSubscribers || HasAllPersistenceIdSubscribers)
@@ -417,10 +422,10 @@ protected override async Task<IImmutableList<Exception>> WriteMessagesAsync(IEnu
417422
eventTagsBatch.Add(new TableTransactionAction(TableTransactionActionType.UpsertReplace, item.WriteEntity()));
418423
}
419424

420-
var eventTagsResponse = await Table.SubmitTransactionAsync(eventTagsBatch, _shutdownCts.Token);
425+
var eventTagsResponse = await Table.ExecuteBatchAsLimitedBatches(eventTagsBatch, _shutdownCts.Token);
421426

422427
if (_log.IsDebugEnabled && _settings.VerboseLogging)
423-
foreach (var r in eventTagsResponse.Value)
428+
foreach (var r in eventTagsResponse)
424429
_log.Debug("Azure table storage wrote entity with status code [{0}]", r.Status);
425430

426431
if (HasTagSubscribers && taggedEntries.Count != 0)

0 commit comments

Comments
 (0)