AMQP - RabbitMQ inputs fail with inequivalent arg 'x-queue-type' errors

[vilitux]

After upgrading RabbitMQ to 3.8.1 and defining a new queue, Graylog is no longer able to consume this new queue. When trying to spawn a new input, the following error is shown in the logs:

2019-11-19 13:28:05,651 ERROR: org.graylog2.shared.inputs.InputLauncher - The [org.graylog2.inputs.syslog.amqp.SyslogAMQPInput] input with ID <5dd3ed3b7f7d630014c6f355> misfired. Reason: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'my-queue' in vhost '/': received none but current is the value 'classic' of type 'longstr', class-id=50, method-id=10).

Expected Behavior

Graylog successfully consumes messages from RabbitMQ running on 3.8.x.

Current Behavior

When spawning a new AMQP input, we're receiving this error:

2019-11-19 13:28:05,651 ERROR: org.graylog2.shared.inputs.InputLauncher - The [org.graylog2.inputs.syslog.amqp.SyslogAMQPInput] input with ID <5dd3ed3b7f7d630014c6f355> misfired. Reason: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'my-queue' in vhost '/': received none but current is the value 'classic' of type 'longstr', class-id=50, method-id=10).

Possible Solution

Make x-queue-type configurable/supplyable.

Steps to Reproduce (for bugs)

1. make sure you have RabbitMQ 3.8.x running
2. create a new queue, where the definition would be something along the following:

    {
      "name": "my-queue",
      "vhost": "/",
      "durable": true,
      "auto_delete": false,
      "arguments": {
        "x-queue-type": "classic"
      }
    }

Your Environment

4 node Graylog cluster on Kubernetes. Avg troughput is 4k messages p/s, using only AMQP inputs.

• Graylog Version: 3.1.3 (docker image, graylog:3.1.3)
• RabbitMQ Version: 3.8.1 (docker image, rabbitmq:3.8.1-management)

[zzet]

UPD for issue:

graylog2-server/graylog2-server/src/main/java/org/graylog2/inputs/transports/AmqpConsumer.java

Line 105 in f74a68f

channel.queueDeclare(queueName, true, false, false, null);

channel.queueDeclare(queueName, true, false, false, null);

vs

channel.queueDeclare(queueName, true, true, false, null);

Will be great to have possibility to configure third argument (all of them?)