Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add failure handling for BatchingSqlJournal SelectCurrentPersistenceIds message batching #5094

Merged
merged 3 commits into from
Jun 15, 2021
Merged

Add failure handling for BatchingSqlJournal SelectCurrentPersistenceIds message batching #5094

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8167211
Add failure handling for BatchingSqlJournal SelectCurrentPersistenceI…
Arkatufus Jun 15, 2021
08e3fed
Do not throw an exception inside a failure handler, throwing will swa…
Arkatufus Jun 15, 2021
6e3532c
Add missing using to make sure reader is properly disposed
Arkatufus Jun 15, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
using System.Diagnostics;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Runtime.ExceptionServices;
using System.Runtime.Serialization;
using System.Text;
using System.Threading.Tasks;
Expand Down Expand Up @@ -800,8 +801,15 @@ private void FailChunkExecution(ChunkExecutionFailure message)
replayAll.ReplyTo.Tell(new EventReplayFailure(cause));
break;

case SelectCurrentPersistenceIds select:
// SqlJournal handled this failure case by using the default PipeTo failure
// handler which sends a Status.Failure message back to the sender.
select.ReplyTo.Tell(new Status.Failure(cause));
break;

default:
throw new Exception($"Unknown persistence journal request type [{request.GetType()}]");
Log.Error(cause, $"Batching failure not reported to original sender. Unknown batched persistence journal request type [{request.GetType()}].");
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Throwing would swallow the original failure cause, create confusing error message, and blow up BatchingSqlJournal.

break;
}
}

Expand Down Expand Up @@ -1096,10 +1104,12 @@ protected virtual async Task HandleSelectCurrentPersistenceIds(SelectCurrentPers
command.Parameters.Clear();
AddParameter(command, "@Ordering", DbType.Int64, message.Offset);

var reader = await command.ExecuteReaderAsync();
while (await reader.ReadAsync())
using (var reader = await command.ExecuteReaderAsync())
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Need to add using to make sure that the DataReader is properly closed and disposed. Failure to dispose the DataReader can cause rollback failure when the batch needs to be rolled back.

{
result.Add(reader.GetString(0));
while (await reader.ReadAsync())
{
result.Add(reader.GetString(0));
}
}

message.ReplyTo.Tell(new CurrentPersistenceIds(result, highestOrderingNumber));
Expand Down