AmqpRejectAndDontRequeueException completely restarts Consumer about spring-amqp HOT 2 CLOSED

new Thread(() -> {

long tagToRollback = isAsyncReplies()
							? message.getMessageProperties().getDeliveryTag()
							: -1;

		boolean ackRequired = !this.acknowledgeMode.isAutoAck()
				&& (!this.acknowledgeMode.isManual() || ContainerUtils.isRejectManual(ex));
...
if (ackRequired) {
				if (tag < 0) {
					OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
					if (deliveryTag.isPresent()) {
						this.channel.basicNack(deliveryTag.getAsLong(), true,
								ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
					}

    /**
     * Reject one or several received messages.
     *
     * Supply the <code>deliveryTag</code> from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
     * or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method containing the message to be rejected.
     * @see com.rabbitmq.client.AMQP.Basic.Nack
     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
     * @param multiple true to reject all messages up to and including
     * the supplied delivery tag; false to reject just the supplied
     * delivery tag.
     * @param requeue true if the rejected message(s) should be requeued rather
     * than discarded/dead-lettered
     * @throws java.io.IOException if an error is encountered
     */
    void basicNack(long deliveryTag, boolean multiple, boolean requeue)

   @RabbitListener(queues = "demo-queue", containerFactory = "demoRabbitConnectionFactory")
    public CompletableFuture<Void> demoListener(String payload, @Header(AmqpHeaders.DELIVERY_TAG) final long tag) {
        return CompletableFuture.runAsync(() -> {
            log.info("Received message with tag: {}", tag);

            try {
                log.info("Started processing message with tag: {}", tag);
                Thread.sleep(2000L); // simulate some off-line processing
                log.info("Finished processing message with tag: {}", tag);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            log.info("Accepted message {} with tag: {}", payload, tag);
        });
    }