SubscriberNotFoundException when I run each subscriber in a separated process

[rafaelpadovezi]

Hi,

I have a specific scenario and would like some guidance on whether this is possible. I'm running CAP with subscribers distributed across separate processes (different pods). When a process receives a message from the database but doesn't have the corresponding subscriber registered, it throws a SubscriberNotFoundException.

In my setup, this is expected behavior since subscribers are intentionally separated. I'd prefer to simply ignore these messages when no subscriber is found rather than throwing an exception. If I'm not mistaken, this exception "short-circuits" the message retries according to the code.

Is there a configuration option or way to disable this exception and silently skip unmatched messages?

[yang-xiaodong]

When a process receives a message from the database but doesn't have the corresponding subscriber registered

Can you explain why this happened? When the same App runs multiple instance, you can enable the UseStorageLock option to solve the concurrency problem.

[rafaelpadovezi]

I have multiple instances of the same app running, but I implemented a logic to register only 1 subscriber per instance. I did this because I want to be able to scale each consumer individually. So, I have multiple instances of the app sharing the same DB (because they are the same micro service) but 1 subscriber per process (or pod, since I use kubernetes).

When a subscriber gets a message from the DB to retry, It can get a message of a subscriber that is not registered in this instance so it will throw a SubscriberNotFoundException and stop retrying this message.

I wish CAP only try to retry messages from subscribers registered (what I believe is way too complicated).

Another thing that would work for me is to ignore the message if no subscriber is found for the message. This means not throwing the SubscriberNotFoundException and specially not setting the retry count as the max retry.

I know that is not the correct use case for CAP, but I hope to get some guidance to the better approach to fit my architecture.

[yang-xiaodong]

I don’t think this is a good design. If a pod crashes or is evicted, messages that are bound to a pod-specific consumer can become orphaned and may remain unprocessed unless there is manual intervention.

I’m not sure what you mean by “scale each consumer individually.” If the goal is to scale different processing paths independently, you can treat each consumer like an MVC Controller action: split high-throughput and low-throughput handlers into separate deployments (but keep all replicas of the same handler in the same consumer group). This lets you scale each handler independently without creating stranded messages.

[yang-xiaodong]

If you really want to stick with your approach, you could try using the instance’s hostname as the Version option value. CAP includes the version field when pulling retry messages from storage.

[rafaelpadovezi]

I’m not sure what you mean by “scale each consumer individually.” If the goal is to scale different processing paths independently, you can treat each consumer like an MVC Controller action: split high-throughput and low-throughput handlers into separate deployments (but keep all replicas of the same handler in the same consumer group). This lets you scale each handler independently without creating stranded messages.

I was not clear before. Each deployment process one RabbitMQ queue (or group). If a pod dies, and k8s is able to create a new healthy pod, the message will be reprocessed. Is that what you mean? If so, I still have the SubscriberNotFoundException error happening because of the split when the app consumes more than one queue.

If you really want to stick with your approach, you could try using the instance’s hostname as the Version option value. CAP includes the version field when pulling retry messages from storage.

That's an interesting solution. I will try it but using the group name as the version instead of instance hostname.

[yang-xiaodong]

Do you configure different services with the same table? Do you know how SubscriberNotFoundException is generated?

[rafaelpadovezi]

Do you know how SubscriberNotFoundException is generated?

I'm supposing some stuff here just looking at CAP code and the exception logged.

1. When getting the messages for retrying, this query is executed on the db, without filtering by message type.
2. Since the Subscriber with group name of the message is not registered in the DI, the code will enter here and then the method SetFailedState will set the message.Retries = _options.FailedRetryCount;

If this is not the expected behavior I will go deeper in my analysis. Please let me know.

Do you configure different services with the same table?

Yes, same table. I was considering different tables but I imagine this will affect how we use the cap dashboard (each deployment would have to have a dashboard). So I'm trying the Version solution because it looks like a better alternative. But I still wasn't capable of making it work because, on RabbitMQ, the exchange name uses the version.

Thank you so much for your help @yang-xiaodong

[yang-xiaodong]

You should use different tables for different services. Use DotNetCore.CAP.Dashboard.K8S package to integrate the dashboard.

[rafaelpadovezi]

I will consider different tables, but the version would be a better fit for my use case if setting a version didn't change the Exchange name. What is the reason for this?

[yang-xiaodong]

Different versions may introduce substantial changes that impact consumers. In particular, certain topics might be removed, and parameter definitions may be modified, which could lead to incompatibility issues.

• Forward Compatibility: Although different queues can be used to distinguish versions, our previous design already follows this approach. Therefore, it is necessary to maintain compatibility with the existing model.
• Migration Path: A complete migration ensures that the new version can be processed efficiently without residual dependencies.
• Message Isolation: The design guarantees that services running on the new version will not inadvertently receive messages originating from the old version, preventing cross-version conflicts.

[rafaelpadovezi]

Hi @yang-xiaodong,

Thanks so much for your help. Now I have 2 options:

• The table name it's a solid solution. The only problem I have is with CAP Dashboard and, although the K8S.Dashboard package seems to solve the problem, it will require more work on my existing services
• Versioning already solves my problem with less impact. But, as I said before, changing the exchange name per version is a deal breaker.

Can I open a PR to allow keeping the exchange name as set in the config? A new option in UseRabbitMQ or at least allowing to manipulate Exchange property in a derived class should do the trick. It seems to have more people wanting this

[yang-xiaodong]

Have you fully validated your idea on other brokers that do not have Exchange, such as Kafka? For example, does re-executing on the Dashboard correctly route to the right consumer? I think this might not work correctly.

[rafaelpadovezi]

I was thinking in a solution specific to RabbitMQ, because the ExchangeName property is set when calling UseRabbitMQ. But honestly, I was not capable to think in a good solution for this problem. You can close the issue. Thanks @yang-xiaodong.