MassTransit not subscribing to AzureServiceBus Topic-CodePudding

I'm currently trying to update application that was originally .NET Core 3.1 using MassTransit 6.3.2. It is now configured to use .NET 6.0 and MassTransit 7.3.0

Our application uses MassTransit to send messages via Azure Service Bus, publishing messages to Topics, which then have other Subscribers listening to those Topic.

Cut down, it was implemented like so:

// Program.cs
services.AddMassTransit(config =>
{
   config.AddConsumer<AppointmentBookedMessageConsumer>();
   config.AddBus(BusControlFactory.ConfigureAzureServiceBus);
});


// BusControlFactory.cs
public static class BusControlFactory
{
   public static IBusControl ConfigureAzureServiceBus(IRegistrationContext<IServiceProvider> context)
   {
      var config = context.Container.GetService<AppConfiguration>();
      var azureServiceBus = Bus.Factory.CreateUsingAzureServiceBus(busFactoryConfig =>
      {
         busFactoryConfig.Host("Endpoint=sb://REDACTED-queues.servicebus.windows.net/;SharedAccessKeyName=MyMessageQueuing;SharedAccessKey=MyKeyGoesHere");
         busFactoryConfig.Message<AppointmentBookedMessage>(m => m.SetEntityName("appointment-booked"));
         busFactoryConfig.SubscriptionEndpoint<AppointmentBookedMessage>(
            "my-subscriber-name",
            configurator =>
            {
               configurator.UseMessageRetry(r => r.Interval(5, TimeSpan.FromSeconds(60)));
               configurator.Consumer<AppointmentBookedMessageConsumer>(context.Container);
            });
         return azureServiceBus;
      }
   }
}

It has now been changed and upgraded to the latest MassTransit and is implemented like:

// Program.cs
services.AddMassTransit(config =>
{
   config.AddConsumer<AppointmentBookedMessageConsumer, AppointmentBookedMessageConsumerDefinition>();
   config.UsingAzureServiceBus((context, cfg) => 
   {
      cfg.Host("Endpoint=sb://REDACTED-queues.servicebus.windows.net/;SharedAccessKeyName=MyMessageQueuing;SharedAccessKey=MyKeyGoesHere");
      cfg.Message<AppointmentBookedMessage>(m => m.SetEntityName("appointment-booked"));

      cfg.ConfigureEndpoints(context);
});


// AppointmentBookedMessageConsumerDefinition.cs
public class AppointmentBookedMessageConsumerDefinition: ConsumerDefinition<AppointmentBookedMessageConsumer>
{
   public AppointmentBookedMessageConsumerDefinition()
   {
      EndpointName = "testharness.subscriber";
   }

   protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<AppointmentBookedMessageConsumer> consumerConfigurator)
   {
      endpointConfigurator.UseMessageRetry(r => r.Interval(5, TimeSpan.FromSeconds(60)));
   }
}

The issue if it can be considered one, is that I can't bind to a subscription that already exists.

In the example above, you can see that the EndpointName is set as "testharness.subscriber". There was already a subscription to the Topic "appointment-booked" from prior to me upgrading. However, when the application runs, it does not error, but it receives no messages.

If I change the EndpointName to "testharness.subscriber2". Another subscriber appears in the Azure Service Bus topic (via the Azure Portal) and I start receiving messages. I can see no difference in the names (other than the change that I placed, in this case: the "2" suffix).

Am I missing something here? Is there something else I need to do to get these to bind? Is my configuration wrong? Was it wrong? While I'm sure I can get around this by managing the release more closely and removing unneeded queues once they're using new ones - it feels like the wrong approach.

CodePudding user response：

With Azure Service Bus, ForwardTo on a subscription can be a bit opaque.

While the subscription may indeed visually indicate that it is forwarding to the correctly named queue, it might be that the queue was deleted and recreated at some point without deleting the subscription. This results in a subscription that will build up messages, as it is unable to forward them to a queue that no longer exists.

Why? Internally, a subscription maintains the ForwardTo as an object id, which after the queue is deleted points to an object that doesn't exist – resulting in messages building up in the subscription.

If you have messages in the subscription, you may need to go into the portal and update that subscription to point to the new queue (even though it has the same name), at which point the messages should flow through to the queue.

If there aren't any messages in the subscription (or if they aren't important), you can just delete the subscription and it will be recreated by MassTransit when you restart the bus.