RabbitMQ channels idling issue | How to recover unacked AMQP messages | Javaclient consumer

Question

I am a consumer for rabbitmq and using spring-amqp. Right now when i go into admin, all the connections are showing running but channels inside them are all idle (Prefetch:250, Unacked:250) . Could you Please help? How to use this prefetch properly ? Do i need to close connections ? How can I increase channels per connection . Right now there is only one channel per connection. Following is the code configuration snippet. I am using out if the box spring amqp configuration for most of the things .Also I am using a custom rabbitmq message listener to ack or unack messages.

<!-- RabbitMQ configuration -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.vhosts}" requested-heartbeat="${rabbitmq.requestedHeartBeat}"/>

    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" message-converter="rabbitJsonConverter"/>
    channel.Close
    <bean id="rabbitJsonConverter" class="rabbitmq.messages.converter.CustomJackson2JsonMessageConverter">
        <property name="classMapper">
            <bean class="org.springframework.amqp.support.converter.DefaultClassMapper">
                <property name="defaultType" value="rabbitmq.messages.custom.dto.CustomRabbitMQMessage"/>
            </bean>
        </property>
    </bean>

    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" requeue-rejected="true">
        <rabbit:listener queue-names="${rabbitmq.queuename}" ref="customRabbitMQMessageListener" method="onMessage"/>
    </rabbit:listener-container>

<bean id="customRabbitMQMessageListener" class="rabbitmq.messages.listener.CustomRabbitMQMessageListener" >
        <property name="customerAccountService" ref="customerAccountService" />
    </bean>

**Listener Code**
LOG.debug("***** LISTENING RABBITMQ MESSAGES START******");
        channel.basicRecover(true);
        try {
                boolean ack = performOperationsOnMessage(msg);
                if (ack) {
                    channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
                } else
                    channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
                LOG.debug("***** LISTENING RABBITMQ MESSAGES FINISHED******");
        } catch (Exception exp) {
            LOG.error("Exception occured during perform Change operation, RabbitMQ message: " + exp.getMessage(), exp);
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
        }

private boolean performOperationsOnMessage(Message msg) {
        RabbitMQMessage message = null;
            try {
                message = (RabbitMQMessage) rabbitJsonConverter.fromMessage(msg);
            } catch (MessageConversionException exp) {
                LOG.warn("Exception occurred during the conversion or any other issue", exp);
                return true;
            }
        if (message == null || message.getOperation() == null || message.getResource() == null || message.getResource().getUuid() == null) {
            LOG.warn("Received an empty message  or emptry operation or empty resource or empty uuid from queue ");
            return true;
        }
        if (message.getOperation().equals(RabbitMQMessage.RossoOperation.remove.name())) {
            return performRemoveOperation(message);
        }
        if (message.getOperation().equals(RabbitMQMessage.RossoOperation.change.name())) {
            return performChangeOperation(message);
        }
        return true;
    }

Answer 1

So the solution for this problem was in the listener code that was configured for manual acknowledgement. There were some branches in the logic which were leaving the listener unable to acknowledge some messages and that's how unacked count on the channels reached prefetch (250) leaving RabbitMQ stopping sending messages to the channels.

Fix: As you would see the updated listener code in the question, it never leaves any message unacknowledged. Also in the negative acknowledgement channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true), requeue (last variable in the signature) should be true so that messages can be requeued back to the same queue