Merge pull request #3 from galgus/auto_ack

chore: set auto_ack to true
galgus · Sep 29, 2023 · ac7aed2 · ac7aed2
2 parents d2f3c8b + 5b1b7e4
commit ac7aed2
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 5 deletions.
diff --git a/pom.xml b/pom.xml
@@ -8,7 +8,7 @@
     </parent>
     <groupId>com.galgus.kafka.connect</groupId>
     <artifactId>kafka-connect-rabbitmq</artifactId>
-    <version>0.2.0</version>
+    <version>0.3.0</version>
     <name>kafka-connect-rabbitmq</name>
     <description>A Kafka Connect connector reading and writing data from RabbitMQ.</description>
     <url>https://github.com/galgus/kafka-connect-rabbitmq</url>

diff --git a/src/main/java/com/github/galgus/kafka/connect/rabbitmq/source/RabbitMQSourceTask.java b/src/main/java/com/github/galgus/kafka/connect/rabbitmq/source/RabbitMQSourceTask.java
@@ -21,7 +21,7 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import org.apache.kafka.connect.errors.ConnectException;
-import org.apache.kafka.connect.errors.RetriableException;
+//import org.apache.kafka.connect.errors.RetriableException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
 import org.slf4j.Logger;
@@ -93,7 +93,7 @@ public void start(Map<String, String> settings) {
     for (String queue : config.queues) {
       try {
         log.info("Starting consumer");
-        this.channel.basicConsume(queue, consumer);
+        this.channel.basicConsume(queue, true, consumer);
         log.info("Setting channel.basicQos({}, {});", config.prefetchCount, config.prefetchGlobal);
         this.channel.basicQos(config.prefetchCount, config.prefetchGlobal);
       } catch (IOException ex) {
@@ -103,15 +103,15 @@ public void start(Map<String, String> settings) {
 
   }
 
-  @Override
+  /*@Override
   public void commitRecord(SourceRecord record) {
     Long deliveryTag = (Long) record.sourceOffset().get("deliveryTag");
     try {
       this.channel.basicAck(deliveryTag, false);
     } catch (IOException e) {
       throw new RetriableException(e);
     }
-  }
+  }*/
 
   @Override
   public List<SourceRecord> poll() throws InterruptedException {