Azure Service Bus Topic发布消息无报错但未持久化问题排查求助

2021年10月20日更新 #

向队列发布消息功能运行正常，发布到队列的消息会被持久化存储，但向主题发布的消息不会被持久化。

更新 #

下文新增了可复现相同问题的简单控制台应用代码。

我尝试从Azure Function向Service Bus的主题发送消息，分别测试了两种方案：一是通过Mass Transit使用托管身份发送，二是通过Azure.Messaging.ServiceBus NuGet包使用共享访问密钥发送。两种方案执行都无异常抛出，但主题中始终没有对应消息。

从函数侧发送消息时观测到的指标如下：

主题的附加配置如下：

我可以通过Azure门户内的Service Bus Explorer成功向该主题写入消息。
当前该主题下没有任何订阅，此前测试时曾创建过一个订阅，现已删除。

Mass Transit 配置（Startup.cs） #

private void ConfigureMassTransit(IServiceCollection services, IConfiguration config) {
    const string KEY_QUEUE_SERVER  = "REDACTED";
    const string EMAIL_RETRY_TOPIC = "REDACTED";
    const string EMAIL_SENT_TOPIC  = "REDACTED";

    services.AddMassTransit(x => {
        x.UsingAzureServiceBus((context, cfg) => {
            cfg.Host(new Uri(config[KEY_QUEUE_SERVER]), host => {
                host.TokenProvider = TokenProvider.CreateManagedIdentityTokenProvider();
            });
            cfg.Message<EmailSentEvent>(m => m.SetEntityName(EMAIL_SENT_TOPIC));
            cfg.Message<TransactionEmailFailedEvent>(m => m.SetEntityName(EMAIL_RETRY_TOPIC));

            cfg.ConfigureEndpoints(context);
        });
    });
}

MassTransitQueueAdapter.cs #

public class MassTransitQueueAdapter : IQueueAdapter {
    #region attributes
    private readonly IBus _bus;
    #endregion

    #region ctor
    public MassTransitQueueAdapter(IBus bus) {
        _bus = bus;
    }
    #endregion

    #region methods
    public void PublishFailure(TransactionEmailFailedEvent failedEvent) {
        _bus.Publish(failedEvent);
    }

    public void PublishSuccess(EmailSentEvent sentEvent) {
        _bus.Publish(sentEvent);
    }
    #endregion
}

ServiceBusQueueAdapter.cs #

public class ServiceBusQueueAdapter : IQueueAdapter {
    #region attributes
    private readonly QueueContext _context;
    #endregion

    #region ctor
    public ServiceBusQueueAdapter(QueueContext context) {
        _context = context;
    }
    #endregion

    #region methods
    private static ServiceBusClient BuildClient(string connectionString) => new ServiceBusClient(connectionString);

    public void PublishFailure(TransactionEmailFailedEvent failedEvent) {
        throw new System.NotImplementedException();
    }

    public void PublishSuccess(EmailSentEvent sentEvent) {
        ServiceBusClient client = BuildClient(_context.SentTopicConnectionString);
        ServiceBusSender sender = client.CreateSender(_context.SentTopicName);

        Task.Run(() => sender.SendMessageAsync(new ServiceBusMessage(JsonConvert.SerializeObject(sentEvent))));
    }
    #endregion
}

简单控制台应用代码 #

class Program {
    static void Main() {
        string           cs     = "Endpoint=sb://REDACTED.servicebus.windows.net/;SharedAccessKeyName=test_with_manage;SharedAccessKey=REDACTED;";
        ServiceBusClient client = new ServiceBusClient(cs);
        ServiceBusSender sender = client.CreateSender("test_1");

        sender.SendMessageAsync(new ServiceBusMessage("Hello World!"))
              .Wait();
    }
}

内容的提问来源于stack exchange，提问作者fizch