A build that doesn't use transactions, avoiding empty offsets

build.sh

@@ -21,10 +21,10 @@ set -e
 docker build -f ./Dockerfile -t yolean/kafka-topics-copy:dev .
 
 build-contract
-docker tag yolean/kafka-topics-copy:dev yolean/kafka-topics-copy:latest
-docker push yolean/kafka-topics-copy:latest
+docker tag yolean/kafka-topics-copy:dev yolean/kafka-topics-copy:notx
+docker push yolean/kafka-topics-copy:notx
 
 # Workaround for https://github.com/Yolean/kafka-topics-copy/issues/4
 # TODO add to build-contract
-docker build -f ./Dockerfile --target runtime-plainjava -t yolean/kafka-topics-copy:plainjava .
-docker push yolean/kafka-topics-copy:plainjava
+docker build -f ./Dockerfile --target runtime-plainjava -t yolean/kafka-topics-copy:plainjava-notx .
+docker push yolean/kafka-topics-copy:plainjava-notx

src/main/java/se/yolean/kafka/topicscopy/tasks/CopyByPoll.java

@@ -72,7 +72,7 @@ public void run() {
     }
 
     try {
-      producer.beginTransaction();
+      // producer.beginTransaction();
 
       List<Future<RecordMetadata>> sent = new ArrayList<>(count);
       Iterator<ConsumerRecord<byte[], byte[]>> records = polled.iterator();
@@ -88,33 +88,25 @@ public void run() {
         metadata.add(i, m);
       }
 
-      // https://hevodata.com/blog/kafka-exactly-once/, but what do we send for the cross-cluster mirror case?
-      // producer.sendOffsetsToTransaction(offsets, consumerGroupId);
-      producer.commitTransaction();
+      // producer.commitTransaction();
 
-      // https://www.baeldung.com/kafka-exactly-once says:
-      // Conversely, applications that must read and write to different Kafka clusters
-      // must use the older commitSync and commitAsync API. Typically, applications
-      // will store consumer offsets into their external state storage to maintain
-      // transactionality.
       consumer.commitSync();
 
       statusHandlers.forEach(h -> h.copied(count));
 
     } catch (ProducerFencedException e) {
-      // https://hevodata.com/blog/kafka-exactly-once/ doesn't abortTransaction here
       throw new RuntimeException("Unhandled", e);
     } catch (KafkaException e) {
-      producer.abortTransaction();
+      //producer.abortTransaction();
       throw new RuntimeException("Unhandled", e);
     } catch (InterruptedException e) {
-      producer.abortTransaction();
+      //producer.abortTransaction();
       throw new RuntimeException("Unhandled", e);
     } catch (ExecutionException e) {
-      producer.abortTransaction();
+      //producer.abortTransaction();
       throw new RuntimeException("Unhandled", e);
     } catch (TimeoutException e) {
-      producer.abortTransaction();
+      //producer.abortTransaction();
       throw new RuntimeException("Unhandled", e);
     }
   }

src/main/java/se/yolean/kafka/topicscopy/tasks/Create.java

@@ -32,7 +32,6 @@ public void run() {
     consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, options.getGroupId());
     consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.getSourceBootstrap());
     consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
-    consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
     consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
     consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
     consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, options.getAutoOffsetReset());
@@ -41,11 +40,7 @@ public void run() {
 
     Properties producerProps = new Properties();
     producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, options.getTargetBootstrap());
-    producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
-    // https://www.baeldung.com/kafka-exactly-once
-    // "All that we need to do is make sure the transaction id is distinct for each
-    // producer, though consistent across restarts."
-    producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, options.getGroupId() + "-tx1");
+    producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
     producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
     producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
     producerProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, options.getTargetCompression().name);

src/main/java/se/yolean/kafka/topicscopy/tasks/Subscribe.java

@@ -18,7 +18,7 @@ public Create getCreated() {
 
   @Override
   public void run() {
-    this.created.getProducer().initTransactions();
+    //this.created.getProducer().initTransactions();
 
     this.created.getConsumer().subscribe(sourceTopics);
   }