NIFI-15892 Add negative acknowledgement handling for ConsumeAMQP batches

nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/AMQPConsumer.java

@@ -103,18 +103,30 @@ public GetResponse consume() {
         return responseQueue.poll();
     }
 
-    public void acknowledge(final GetResponse response) {
+    public void acknowledge(final long deliveryTag, final boolean multiple) {
         if (autoAcknowledge) {
             return;
         }
 
         try {
-            getChannel().basicAck(response.getEnvelope().getDeliveryTag(), true);
+            getChannel().basicAck(deliveryTag, multiple);
         } catch (Exception e) {
             throw new AMQPException("Failed to acknowledge message", e);
         }
     }
 
+    public void negativeAcknowledge(final long deliveryTag, final boolean multiple, final boolean requeue) {
+        if (autoAcknowledge) {
+            return;
+        }
+
+        try {
+            getChannel().basicNack(deliveryTag, multiple, requeue);
+        } catch (Exception e) {
+            throw new AMQPException("Failed to negatively acknowledge message", e);
+        }
+    }
+
     @Override
     public void close() throws TimeoutException, IOException {
         try {

nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java

@@ -225,8 +225,19 @@ protected void processResource(final Connection connection, final AMQPConsumer c
         }
 
         if (lastReceived != null) {
-            final GetResponse finalGetResponse = lastReceived;
-            session.commitAsync(() -> consumer.acknowledge(finalGetResponse), null);
+            final long lastDeliveryTag = lastReceived.getEnvelope().getDeliveryTag();
+
+            session.commitAsync(
+                    () -> consumer.acknowledge(lastDeliveryTag, true),
+                    failure -> {
+                        getLogger().warn(
+                                "ProcessSession commit failed after consuming AMQP messages up to delivery tag {}; negatively acknowledging with requeue",
+                                lastDeliveryTag,
+                                failure
+                        );
+                        consumer.negativeAcknowledge(lastDeliveryTag, true, true);
+                    }
+            );
         }
     }