We've begun the journey of migrating from MassTransit 2.9.5 (with Courier addon) to MassTransit 4.0.1. Things are going slow, but OK and I am excited about the new (to us) functionality... but I've hit a wall with Courier.
Here's a braindead simple Activity that has both execute and compensate behavior:
public class AdditionActivity : Activity<AdditionActivity.Args, AdditionActivity.CompLog>
{
private readonly ILogger _logger;
public AdditionActivity(ILogger logger)
{
_logger = logger;
}
public class Args : CorrelatedBy<Guid>
{
public Guid CorrelationId { get; set; }
public int First { get; set; }
public int Second { get; set; }
}
public class CompLog : CorrelatedBy<Guid>
{
public Guid CorrelationId { get; set; }
public int Result { get; set; }
}
public async Task<ExecutionResult> Execute(ExecuteContext<Args> context)
{
var compLog = new CompLog
{
CorrelationId = context.Arguments.CorrelationId,
Result = context.Arguments.First + context.Arguments.Second
};
_logger.Info($"{nameof(AdditionActivity)}[{context.Arguments.CorrelationId} - Adding [{context.Arguments.First}] and [{context.Arguments.Second}] = [{compLog.Result}]");
await Task.CompletedTask;
return context.Completed(compLog);
}
public async Task<CompensationResult> Compensate(CompensateContext<CompLog> context)
{
_logger.Warn($"{nameof(AdditionActivity)}[{context.Log.CorrelationId}] - Ruh roh... must compensate for add that resulted in [{context.Log.Result}]");
await Task.CompletedTask;
return context.Compensated();
}
}
When I execute a routing slip including this activity, it executes (I see the logs, and can step through the code).. but after return context.Completed(compLog), we get an exception in the console (which is also captured in the error queue message here):
Exchange Procedure.Service.Activities.AdditionActivity.Execute_error
Routing Key
Redelivered ●
Properties
message_id: 4ded0000-47b0-38d5-3ae8-08d5701751d5
correlation_id: 924899da-7a61-4b80-819f-f66f166c72e4
delivery_mode: 2
headers:
Content-Type: application/vnd.masstransit+json
publishId: 1
MT-Reason: fault
MT-Fault-ExceptionType: System.ArgumentException
MT-Fault-Message: The routingSlip must contain at least one activity log
MT-Fault-Timestamp: 2018-02-09T23:46:20.3953028Z
MT-Fault-StackTrace: at MassTransit.Courier.Hosts.HostCompensateContext`1..ctor(HostInfo host, ConsumeContext`1 context)
at MassTransit.Courier.Hosts.CompensateActivityHost`2.<Send>d__4.MoveNext()
at MassTransit.Courier.Hosts.CompensateActivityHost`2.<Send>d__4.MoveNext()
at GreenPipes.Filters.TeeFilter`1.<Send>d__5.MoveNext()
at GreenPipes.Filters.OutputPipeFilter`2.<GreenPipes-IFilter<TInput>-Send>d__6.MoveNext()
at GreenPipes.Filters.OutputPipeFilter`2.<GreenPipes-IFilter<TInput>-Send>d__6.MoveNext()
at MassTransit.Pipeline.Filters.DeserializeFilter.<Send>d__4.MoveNext()
at GreenPipes.Filters.RescueFilter`2.<GreenPipes-IFilter<TContext>-Send>d__5.MoveNext()
MT-Host-MachineName: TYLER-XPS13
MT-Host-ProcessName: Procedure.Service.vshost
MT-Host-ProcessId: 13040
MT-Host-Assembly: Procedure.Service
MT-Host-AssemblyVersion: 1.0.0.0
MT-Host-MassTransitVersion: 4.0.1.1390
MT-Host-FrameworkVersion: 4.0.30319.42000
MT-Host-OperatingSystemVersion: Microsoft Windows NT 6.2.9200.0
content_type: application/vnd.masstransit+json
Payload
1844 bytes
Encoding: string
{
"messageId": "4ded0000-47b0-38d5-3ae8-08d5701751d5",
"correlationId": "924899da-7a61-4b80-819f-f66f166c72e4",
"conversationId": "4ded0000-47b0-38d5-a350-08d570175173",
"initiatorId": "924899da-7a61-4b80-819f-f66f166c72e4",
"sourceAddress": "rabbitmq://localhost/Procedure.Service",
"destinationAddress": "rabbitmq://localhost/Procedure.Service.Activities.AdditionActivity.Execute",
"messageType": [
"urn:message:MassTransit.Courier.Contracts:RoutingSlip"
],
"message": {
"trackingNumber": "924899da-7a61-4b80-819f-f66f166c72e4",
"createTimestamp": "2018-02-09T23:46:20.1847429Z",
"itinerary": [
{
"name": "AdditionActivity",
"address": "rabbitmq://localhost/Procedure.Service.Activities.AdditionActivity.Execute",
"arguments": {
"correlationId": "00000000-0000-0000-0000-000000000000",
"first": 3,
"second": 5
}
},
{
"name": "LogActivity",
"address": "rabbitmq://localhost/Procedure.Service.Activities.LogActivity.Execute",
"arguments": {
"theLogMessage": "NewStyleConsumer - Consumed SomeFunMessage[924899da-7a61-4b80-819f-f66f166c72e4] - Procedure [924899da-7a61-4b80-819f-f66f166c72e4 has begun... - gonna trigger a routing slip!"
}
}
],
"activityLogs": [],
"compensateLogs": [],
"variables": {},
"activityExceptions": [],
"subscriptions": []
},
"headers": {},
"host": {
"machineName": "TYLER-XPS13",
"processName": "Procedure.Service.vshost",
"processId": 13040,
"assembly": "Procedure.Service",
"assemblyVersion": "1.0.0.0",
"frameworkVersion": "4.0.30319.42000",
"massTransitVersion": "4.0.1.1390",
"operatingSystemVersion": "Microsoft Windows NT 6.2.9200.0"
}
}
I've read through the documentation here (http://masstransit-project.com/MassTransit/advanced/courier/), but nothing's jumping out at me as wrong (aside from a bit of difference in naming of some things between the doc's code samples and current MT 4.0.1).
What is the proper way to store a compensation log entry on completion of an activity in MT 4.0.1? and/or is there something else I might be doing wrong that is manifesting itself as this exception even though I am providing a compensation log value?
Thanks,
Tyler
It looks like you have the CompensateHost registered on the same endpoint as the execute activity host, which is causing both hosts to be executed when the routing slip is received.
The ExecuteActivityHost and CompensateActivityHost must be on separate endpoints (which I'm guessing you already know, but have somehow wired them up incorrectly after porting to 4.0.1).