RabbitMQ Topology Operator Custom Delayed Message Exchange

[danielshiplett]

Is there a way to create a delayed message exchange using the RabbitMQ Topology Operator?

I have successfully created a custom RabbitMQ Server image with the delayed message plugin installed and enabled. However, when I try to create my delay exchange, I get an error "Declare exchange error: Invalid argument, 'x-delayed-type' must be an existing exchange type".

Here is my Exchange definition:

---
apiVersion: rabbitmq.com/v1beta1
kind: Exchange
metadata:
  name: ace-ios-delay-exchange
spec:
  name: "ace-ios.delay"
  type: x-delayed-message
  autoDelete: false
  durable: true
  arguments:
    delayed-type: direct
  rabbitmqClusterReference:
    name: ace-ios

I believe, based on how definitions and arguments work on other topology resources, that the exchange options are analogous to how I was able to create this exchange in my Java/Spring code:

    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE_DELAY, "x-delayed-message", true, false, args);
    }

[michaelklishin]

The message suggests that the plugin was not enabled. If it was not on the code path, you'd get an exception at the time of enabling it.

Although I have not tried listing a missing plugin in the enabled plugins file, I assume that should fail with an exception as well.

[danielshiplett]

I thought so as well, so I checked the startup logs. It appears that the plugin was started:

2023-06-29 21:55:20.679923+00:00 [info] <0.230.0> Setting cluster name to 'ace-ios' as configured
2023-06-29 21:55:20.691364+00:00 [info] <0.230.0> Running boot step direct_client defined by app rabbit
2023-06-29 21:55:20.691562+00:00 [info] <0.230.0> Running boot step rabbit_maintenance_mode_state defined by app rabbit
2023-06-29 21:55:20.691615+00:00 [info] <0.230.0> Creating table rabbit_node_maintenance_states for maintenance mode status
2023-06-29 21:55:20.691836+00:00 [info] <0.230.0> Running boot step rabbit_management_load_definitions defined by app rabbitmq_management
2023-06-29 21:55:20.692008+00:00 [info] <0.493.0> Resetting node maintenance status
2023-06-29 21:55:20.704716+00:00 [info] <0.552.0> Management plugin: HTTP (non-TLS) listener started on port 15672
2023-06-29 21:55:20.704830+00:00 [info] <0.580.0> Statistics database started.
2023-06-29 21:55:20.704898+00:00 [info] <0.579.0> Starting worker pool 'management_worker_pool' with 3 processes in it
2023-06-29 21:55:20.711736+00:00 [info] <0.595.0> Peer discovery: node cleanup is disabled
2023-06-29 21:55:20.713552+00:00 [info] <0.608.0> Prometheus metrics: HTTP (non-TLS) listener started on port 15692
2023-06-29 21:55:20.713653+00:00 [info] <0.493.0> Ready to start client connection listeners
2023-06-29 21:55:20.715449+00:00 [info] <0.652.0> started TCP listener on [::]:5672
 completed with 7 plugins.
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0> Server startup complete; 7 plugins started.
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_prometheus
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_peer_discovery_k8s
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_delayed_message_exchange
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_peer_discovery_common
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_management
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_web_dispatch
2023-06-29 21:55:20.773534+00:00 [info] <0.493.0>  * rabbitmq_management_agent
2023-06-30 17:05:22.207809+00:00 [warning] <0.7737.1> Declare exchange error: Invalid argument, 'x-delayed-type' must be an existing exchange type
2023-06-30 17:05:22.218025+00:00 [warning] <0.7739.1> Declare exchange error: Invalid argument, 'x-delayed-type' must be an existing exchange type

Could it be the format of the parameter? I noticed when I created a Queue Policy that I had to remove the x- from the argument name. However, trying that with the Exchange definition didn't help. It just led to a different error.

Here is the working Queue Policy:

apiVersion: rabbitmq.com/v1beta1
kind: Policy
metadata:
  name: ace-ios-queue-policy
spec:
  name: "ace-ios-queue-policy"
  pattern: "^ace-ios."
  applyTo: "queues"
  definition:
    dead-letter-exchange: "ace-ios.dlx"
  rabbitmqClusterReference:
    name: ace-ios

And here was its equivalent Queue configuration from Java/Spring:

        @Bean
        public Queue queueGroundData() {
            Queue rtn = new Queue(QUEUE_GROUND, true);
            rtn.addArgument("x-dead-letter-exchange", EXCHANGE_DLX);
            return rtn;
        }

Here was the error from RabbitMQ when I tried making this new Exchange have the argument with the x- removed from the argument name:

2023-06-30 17:06:28.431598+00:00 [warning] <0.7806.1> Declare exchange error: unknown exchange type 'delayed-message'

Note: That caused the error to change from a complaint about the argument to a complaint about the type.

[danielshiplett]

Alright, I just went back and tested the one other permutation of the parameters and it worked!

apiVersion: rabbitmq.com/v1beta1
kind: Exchange
metadata:
  name: ace-ios-delay-exchange
spec:
  name: "ace-ios.delay"
  type: x-delayed-message
  autoDelete: false
  durable: true
  arguments:
    x-delayed-type: direct
  rabbitmqClusterReference:
    name: ace-ios

So, in the case of the Exchange, all argument names must include the x-. For Policies, the argument must NOT include the x-. Is this behavior of the Operator or the RabbitMQ Server?

[ChunyiLyu]

Hi @danielshiplett, that's the RabbitMQ Server behavior. Topology Operator does not do parameter checking on spec.arguments for exchanges or queues.