[fix][io] Acknowledge RabbitMQ message after processing the message successfully (#24354)

Author: lhotari

pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java

@@ -96,17 +96,40 @@ public RabbitMQConsumer(RabbitMQSource source, Channel channel) {
         @Override
         public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                 throws IOException {
-            source.consume(new RabbitMQRecord(Optional.ofNullable(envelope.getRoutingKey()), body));
             long deliveryTag = envelope.getDeliveryTag();
-            // positively acknowledge all deliveries up to this delivery tag to reduce network traffic
-            // since manual message acknowledgments are turned on by default
-            this.getChannel().basicAck(deliveryTag, true);
+            source.consume(new RabbitMQRecord(Optional.ofNullable(envelope.getRoutingKey()), body, () -> {
+                // acknowledge this delivery tag to RabbitMQ
+                try {
+                    this.getChannel().basicAck(deliveryTag, false);
+                } catch (IOException e) {
+                    logger.error("Error while acknowledging envelope {}.", envelope, e);
+                }
+            }, () -> {
+                // negatively acknowledge this delivery tag to RabbitMQ
+                try {
+                    this.getChannel().basicNack(deliveryTag, false, true);
+                } catch (IOException e) {
+                    logger.error("Error while negatively acknowledging envelope {}.", envelope, e);
+                }
+            }));
         }
     }
 
     @Data
     private static class RabbitMQRecord implements Record<byte[]> {
         private final Optional<String> key;
         private final byte[] value;
+        private final Runnable ackFunction;
+        private final Runnable nackFunction;
+
+        @Override
+        public void ack() {
+            ackFunction.run();
+        }
+
+        @Override
+        public void fail() {
+            nackFunction.run();
+        }
     }
 }