Repository: skiptests/astraea
Pull request #326 (2022-05-10 07:01:00)

Integrate normal producer with transactional producer
將`TransactionalProducer`整合到`Producer`介面，當使用者執行`send(collection<sender>`時，底層的producer自行決定是否要用transaction。在我們提供的實作中，`builder.buildTransactional`所回傳的producer會自動以transaction的形式來進行所有的資料寫入，反之，`builder.build`所回傳的producer就是一般寫入，因此一次多筆也只會一筆一筆送出而已

將介面合併可以簡化程式碼，同時也方便我們更換producer，不會因為從normal換到transactional就要更改程式碼


Patch:
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 014e8b60..085708ff 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -90,30 +90,26 @@ public class Performance {
     return IntStream.range(0, argument.producers)
         .mapToObj(
             index ->
-                argument.isolation() == Isolation.READ_COMMITTED
-                    ? ProducerExecutor.of(
-                        argument.topic,
-                        argument.transactionSize,
-                        Producer.builder()
+                ProducerExecutor.of(
+                    argument.topic,
+                    // Only transactional producer needs to process batch data
+                    argument.isolation() == Isolation.READ_COMMITTED ? argument.transactionSize : 1,
+                    argument.isolation() == Isolation.READ_COMMITTED
+                        ? Producer.builder()
                             .configs(argument.allConfigs())
                             .brokers(argument.brokers)
                             .compression(argument.compression)
                             .partitionClassName(argument.partitioner)
-                            .buildTransactional(),
-                        observers.get(index),
-                        partitionSupplier,
-                        dataSupplier)
-                    : ProducerExecutor.of(
-                        argument.topic,
-                        Producer.builder()
+                            .buildTransactional()
+                        : Producer.builder()
                             .configs(argument.allConfigs())
                             .brokers(argument.brokers)
                             .compression(argument.compression)
                             .partitionClassName(argument.partitioner)
                             .build(),
-                        observers.get(index),
-                        partitionSupplier,
-                        dataSupplier))
+                    observers.get(index),
+                    partitionSupplier,
+                    dataSupplier))
         .collect(Collectors.toUnmodifiableList());
   }
 
diff --git a/app/src/main/java/org/astraea/performance/ProducerExecutor.java b/app/src/main/java/org/astraea/performance/ProducerExecutor.java
index b584b9f5..12c8f42c 100644
--- a/app/src/main/java/org/astraea/performance/ProducerExecutor.java
+++ b/app/src/main/java/org/astraea/performance/ProducerExecutor.java
@@ -12,14 +12,13 @@ import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.producer.Producer;
 import org.astraea.producer.Sender;
-import org.astraea.producer.TransactionalProducer;
 
 abstract class ProducerExecutor implements Executor {
 
   static ProducerExecutor of(
       String topic,
-      int transactionSize,
-      TransactionalProducer<byte[], byte[]> producer,
+      int batchSize,
+      Producer<byte[], byte[]> producer,
       BiConsumer<Long, Long> observer,
       Supplier<Integer> partitionSupplier,
       DataSupplier dataSupplier) {
@@ -28,7 +27,7 @@ abstract class ProducerExecutor implements Executor {
       @Override
       public State execute() {
         var data =
-            IntStream.range(0, transactionSize)
+            IntStream.range(0, batchSize)
                 .mapToObj(i -> dataSupplier.get())
                 .collect(Collectors.toUnmodifiableList());
 
@@ -64,7 +63,7 @@ abstract class ProducerExecutor implements Executor {
 
       State doSend(List<Sender<byte[], byte[]>> senders) {
         producer
-            .transaction(senders)
+            .send(senders)
             .forEach(
                 future ->
                     future.whenComplete(
@@ -160,7 +159,7 @@ abstract class ProducerExecutor implements Executor {
 
   /** @return true if the producer in this executor is transactional. */
   boolean transactional() {
-    return producer instanceof TransactionalProducer;
+    return producer.transactional();
   }
 
   /** @return true if this executor is closed. otherwise, false */
diff --git a/app/src/main/java/org/astraea/producer/AbstractSender.java b/app/src/main/java/org/astraea/producer/AbstractSender.java
index 7ba1bc21..05577a18 100644
--- a/app/src/main/java/org/astraea/producer/AbstractSender.java
+++ b/app/src/main/java/org/astraea/producer/AbstractSender.java
@@ -3,6 +3,7 @@ package org.astraea.producer;
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.astraea.consumer.Header;
 
 abstract class AbstractSender<Key, Value> implements Sender<Key, Value> {
@@ -48,4 +49,13 @@ abstract class AbstractSender<Key, Value> implements Sender<Key, Value> {
     this.headers = headers;
     return this;
   }
+
+  /**
+   * this helper method is used by our producer only.
+   *
+   * @return a kafka producer record
+   */
+  ProducerRecord<Key, Value> record() {
+    return new ProducerRecord<>(topic, partition, timestamp, key, value, Header.of(headers));
+  }
 }
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index 4d82ff6c..82eabb9d 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -16,7 +16,6 @@ import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.errors.AuthorizationException;
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
 import org.apache.kafka.common.errors.ProducerFencedException;
-import org.astraea.consumer.Header;
 import org.astraea.topic.Compression;
 
 public class Builder<Key, Value> {
@@ -62,7 +61,8 @@ public class Builder<Key, Value> {
   }
 
   private static <Key, Value> CompletionStage<Metadata> doSend(
-      KafkaProducer<Key, Value> producer, ProducerRecord<Key, Value> record) {
+      org.apache.kafka.clients.producer.Producer<Key, Value> producer,
+      ProducerRecord<Key, Value> record) {
     var completableFuture = new CompletableFuture<Metadata>();
     producer.send(
         record,
@@ -75,44 +75,15 @@ public class Builder<Key, Value> {
 
   @SuppressWarnings("unchecked")
   public Producer<Key, Value> build() {
-    var kafkaProducer =
+    return new NormalProducer<>(
         new KafkaProducer<>(
             configs,
             Serializer.of((Serializer<Key>) keySerializer),
-            Serializer.of((Serializer<Value>) valueSerializer));
-    return new Producer<>() {
-
-      @Override
-      public Sender<Key, Value> sender() {
-        return new AbstractSender<>() {
-          @Override
-          public CompletionStage<Metadata> run() {
-            return doSend(
-                kafkaProducer,
-                new ProducerRecord<>(topic, partition, timestamp, key, value, Header.of(headers)));
-          }
-        };
-      }
-
-      @Override
-      public void flush() {
-        kafkaProducer.flush();
-      }
-
-      @Override
-      public KafkaProducer<Key, Value> kafkaProducer() {
-        return kafkaProducer;
-      }
-
-      @Override
-      public void close() {
-        kafkaProducer.close();
-      }
-    };
+            Serializer.of((Serializer<Value>) valueSerializer)));
   }
 
   @SuppressWarnings("unchecked")
-  public TransactionalProducer<Key, Value> buildTransactional() {
+  public Producer<Key, Value> buildTransactional() {
     var transactionConfigs = new HashMap<>(configs);
     transactionConfigs.putIfAbsent(
         ProducerConfig.TRANSACTIONAL_ID_CONFIG, "id" + new Random().nextLong());
@@ -123,82 +94,109 @@ public class Builder<Key, Value> {
             Serializer.of((Serializer<Key>) keySerializer),
             Serializer.of((Serializer<Value>) valueSerializer));
     transactionProducer.initTransactions();
-    return new TransactionalProducer<>() {
-
-      @Override
-      public Sender<Key, Value> sender() {
-        return new TransactionalSender<>() {
-          /** Send one transactional record. */
-          @Override
-          public CompletionStage<Metadata> run() {
-            return transaction(List.of(this)).stream().findFirst().orElseThrow();
-          }
-
-          @Override
-          CompletionStage<Metadata> send() {
-            return doSend(
-                transactionProducer,
-                new ProducerRecord<>(topic, partition, timestamp, key, value, Header.of(headers)));
-          }
-        };
-      }
+    return new TransactionalProducer<>(transactionProducer);
+  }
 
-      @Override
-      public KafkaProducer<Key, Value> kafkaProducer() {
-        return transactionProducer;
-      }
+  private abstract static class BaseProducer<Key, Value> implements Producer<Key, Value> {
+    protected final org.apache.kafka.clients.producer.Producer<Key, Value> kafkaProducer;
 
-      @Override
-      public void flush() {
-        transactionProducer.flush();
-      }
+    private BaseProducer(org.apache.kafka.clients.producer.Producer<Key, Value> kafkaProducer) {
+      this.kafkaProducer = kafkaProducer;
+    }
 
-      @Override
-      public void close() {
-        transactionProducer.close();
-      }
+    @Override
+    public void flush() {
+      kafkaProducer.flush();
+    }
+
+    @Override
+    public void close() {
+      kafkaProducer.close();
+    }
+  }
 
-      /**
-       * Send a collection of records as a transaction operation. For example,
-       *
-       * <pre>{@code
-       * try(var producer = Producer.builder().brokers("localhost:9092").buildTransactional()){
-       *     producer.transaction(
-       *             IntStream.range(0, 10)
-       *                     .mapToObj(i -> producer.sender().topic("topic1").value(new byte[10]))
-       *                     .collect(Collectors.toList()));
-       * }
-       * }</pre>
-       */
-      @Override
-      public Collection<CompletionStage<Metadata>> transaction(
-          Collection<Sender<Key, Value>> senders) {
-        if (!senders.stream().allMatch(s -> s instanceof TransactionalSender))
-          throw new IllegalArgumentException(
-              "Don't pass non-transactional sender: to transactional producer");
-        try {
-          synchronized (transactionProducer) {
-            transactionProducer.beginTransaction();
-            var futures =
-                senders.stream()
-                    .map(s -> ((TransactionalSender<Key, Value>) s).send())
-                    .collect(Collectors.toUnmodifiableList());
-            transactionProducer.commitTransaction();
-            return futures;
-          }
-        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
-          transactionProducer.close();
-          // Error occur
-          throw e;
-        } catch (KafkaException ke) {
-          transactionProducer.abortTransaction();
-          return transaction(senders);
+  private static class NormalProducer<Key, Value> extends BaseProducer<Key, Value> {
+    private NormalProducer(org.apache.kafka.clients.producer.Producer<Key, Value> kafkaProducer) {
+      super(kafkaProducer);
+    }
+
+    @Override
+    public Sender<Key, Value> sender() {
+      return new AbstractSender<>() {
+        @Override
+        public CompletionStage<Metadata> run() {
+          return doSend(kafkaProducer, record());
         }
-      }
-    };
+      };
+    }
+
+    @Override
+    public Collection<CompletionStage<Metadata>> send(Collection<Sender<Key, Value>> senders) {
+      return senders.stream().map(Sender::run).collect(Collectors.toUnmodifiableList());
+    }
+
+    @Override
+    public boolean transactional() {
+      return false;
+    }
   }
 
-  private abstract static class TransactionalSender<Key, Value> extends AbstractSender<Key, Value> {
-    abstract CompletionStage<Metadata> send();
+  private static class TransactionalProducer<Key, Value> extends BaseProducer<Key, Value> {
+    private final Object lock = new Object();
+
+    private TransactionalProducer(
+        org.apache.kafka.clients.producer.Producer<Key, Value> kafkaProducer) {
+      super(kafkaProducer);
+    }
+
+    @Override
+    public Sender<Key, Value> sender() {
+      return new AbstractSender<>() {
+        @Override
+        public CompletionStage<Metadata> run() {
+          return send(List.of(this)).iterator().next();
+        }
+      };
+    }
+
+    @Override
+    public Collection<CompletionStage<Metadata>> send(Collection<Sender<Key, Value>> senders) {
+      var invalidSenders =
+          senders.stream()
+              .filter(s -> !(s instanceof AbstractSender))
+              .collect(Collectors.toUnmodifiableList());
+      if (!invalidSenders.isEmpty())
+        throw new IllegalArgumentException(
+            "those senders: "
+                + invalidSenders.stream()
+                    .map(Sender::getClass)
+                    .map(Class::getName)
+                    .collect(Collectors.joining(","))
+                + " are not supported");
+      try {
+        synchronized (lock) {
+          kafkaProducer.beginTransaction();
+          var futures =
+              senders.stream()
+                  .map(s -> (AbstractSender<Key, Value>) s)
+                  .map(s -> doSend(kafkaProducer, s.record()))
+                  .collect(Collectors.toUnmodifiableList());
+          kafkaProducer.commitTransaction();
+          return futures;
+        }
+      } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
+        kafkaProducer.close();
+        // Error occur
+        throw e;
+      } catch (KafkaException ke) {
+        kafkaProducer.abortTransaction();
+        return send(senders);
+      }
+    }
+
+    @Override
+    public boolean transactional() {
+      return true;
+    }
   }
 }
diff --git a/app/src/main/java/org/astraea/producer/Producer.java b/app/src/main/java/org/astraea/producer/Producer.java
index 11ddbbd0..21ae4b29 100644
--- a/app/src/main/java/org/astraea/producer/Producer.java
+++ b/app/src/main/java/org/astraea/producer/Producer.java
@@ -1,17 +1,29 @@
 package org.astraea.producer;
 
+import java.util.Collection;
 import java.util.Map;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import java.util.concurrent.CompletionStage;
 
 /** An interface for sending records. */
 public interface Producer<Key, Value> extends AutoCloseable {
   Sender<Key, Value> sender();
+
+  /**
+   * send the multiple records. Noted that the normal producer will send the record one by one. By
+   * contrast, transactional producer will send all records in single transaction.
+   *
+   * @param senders pre-defined records
+   * @return callback of all completed records
+   */
+  Collection<CompletionStage<Metadata>> send(Collection<Sender<Key, Value>> senders);
+
   /** this method is blocked until all data in buffer are sent. */
   void flush();
 
   void close();
 
-  KafkaProducer<Key, Value> kafkaProducer();
+  /** @return true if the producer supports transactional. */
+  boolean transactional();
 
   static Builder<byte[], byte[]> builder() {
     return new Builder<>();
diff --git a/app/src/main/java/org/astraea/producer/TransactionalProducer.java b/app/src/main/java/org/astraea/producer/TransactionalProducer.java
deleted file mode 100644
index f2ab8846..00000000
--- a/app/src/main/java/org/astraea/producer/TransactionalProducer.java
+++ /dev/null
@@ -1,17 +0,0 @@
-package org.astraea.producer;
-
-import java.util.Collection;
-import java.util.concurrent.CompletionStage;
-
-/**
- * Difference between this and {@link org.astraea.producer.Producer} is that it is used for sending
- * records as one transaction. TransactionalProducer can be generated from {@link
- * org.astraea.producer.Builder} For example,
- *
- * <pre>{@code
- * var producer = Producer.builder().brokers("localhost:9092").buildTransactional();
- * }</pre>
- */
-public interface TransactionalProducer<Key, Value> extends Producer<Key, Value> {
-  Collection<CompletionStage<Metadata>> transaction(Collection<Sender<Key, Value>> senders);
-}


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
index 604c5a58..909b2a65 100644
--- a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
+++ b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
@@ -22,35 +22,8 @@ import org.junit.jupiter.params.provider.MethodSource;
 
 public class ProducerExecutorTest extends RequireBrokerCluster {
 
-  private static ProducerExecutor ofProducer(
-      String topic,
-      Supplier<Integer> partitionSupplier,
-      BiConsumer<Long, Long> observer,
-      DataSupplier dataSupplier) {
-    return ProducerExecutor.of(
-        topic,
-        Producer.builder().brokers(bootstrapServers()).build(),
-        observer,
-        partitionSupplier,
-        dataSupplier);
-  }
-
-  private static ProducerExecutor ofTransactionalProducer(
-      String topic,
-      Supplier<Integer> partitionSupplier,
-      BiConsumer<Long, Long> observer,
-      DataSupplier dataSupplier) {
-    return ProducerExecutor.of(
-        topic,
-        10,
-        Producer.builder().brokers(bootstrapServers()).buildTransactional(),
-        observer,
-        partitionSupplier,
-        dataSupplier);
-  }
-
   @ParameterizedTest
-  @MethodSource("offsetProducerExecutors")
+  @MethodSource("offerProducerExecutors")
   void testSpecifiedPartition(ProducerExecutor executor) throws InterruptedException {
     var specifiedPartition = 1;
     ((MyPartitionSupplier) executor.partitionSupplier()).partition = specifiedPartition;
@@ -83,21 +56,21 @@ public class ProducerExecutorTest extends RequireBrokerCluster {
   }
 
   @ParameterizedTest
-  @MethodSource("offsetProducerExecutors")
+  @MethodSource("offerProducerExecutors")
   void testDone(ProducerExecutor executor) throws InterruptedException {
     ((MyDataSupplier) executor.dataSupplier()).data = DataSupplier.NO_MORE_DATA;
     Assertions.assertEquals(State.DONE, executor.execute());
   }
 
   @ParameterizedTest
-  @MethodSource("offsetProducerExecutors")
+  @MethodSource("offerProducerExecutors")
   void testClose(ProducerExecutor executor) {
     executor.close();
     Assertions.assertTrue(executor.closed());
   }
 
   @ParameterizedTest
-  @MethodSource("offsetProducerExecutors")
+  @MethodSource("offerProducerExecutors")
   void testObserver(ProducerExecutor executor) throws InterruptedException {
     Assertions.assertEquals(State.RUNNING, executor.execute());
     // wait for async call
@@ -140,22 +113,29 @@ public class ProducerExecutorTest extends RequireBrokerCluster {
     }
   }
 
-  private static Stream<Arguments> offsetProducerExecutors() {
+  private static Stream<Arguments> offerProducerExecutors() {
     var normalTopic = Utils.randomString(10);
     var transactionalTopic = Utils.randomString(10);
     return Stream.of(
         Arguments.of(
             Named.of(
                 "normal producer for topic: " + normalTopic,
-                ofProducer(
-                    normalTopic, new MyPartitionSupplier(), new Observer(), new MyDataSupplier()))),
+                ProducerExecutor.of(
+                    normalTopic,
+                    1,
+                    Producer.builder().brokers(bootstrapServers()).build(),
+                    new Observer(),
+                    new MyPartitionSupplier(),
+                    new MyDataSupplier()))),
         Arguments.of(
             Named.of(
                 "transactional producer for topic: " + transactionalTopic,
-                ofTransactionalProducer(
+                ProducerExecutor.of(
                     transactionalTopic,
-                    new MyPartitionSupplier(),
+                    10,
+                    Producer.builder().brokers(bootstrapServers()).buildTransactional(),
                     new Observer(),
+                    new MyPartitionSupplier(),
                     new MyDataSupplier()))));
   }
 }
diff --git a/app/src/test/java/org/astraea/producer/ProducerTest.java b/app/src/test/java/org/astraea/producer/ProducerTest.java
index 0a422a0e..17177921 100644
--- a/app/src/test/java/org/astraea/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/producer/ProducerTest.java
@@ -4,14 +4,24 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.astraea.Utils;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
 import org.astraea.consumer.Isolation;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Named;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
 
 public class ProducerTest extends RequireBrokerCluster {
 
@@ -23,6 +33,7 @@ public class ProducerTest extends RequireBrokerCluster {
     var header = Header.of("a", "b".getBytes());
     try (var producer =
         Producer.builder().brokers(bootstrapServers()).keySerializer(Serializer.STRING).build()) {
+      Assertions.assertFalse(producer.transactional());
       var metadata =
           producer
               .sender()
@@ -67,6 +78,7 @@ public class ProducerTest extends RequireBrokerCluster {
             .brokers(bootstrapServers())
             .keySerializer(Serializer.STRING)
             .buildTransactional()) {
+      Assertions.assertTrue(producer.transactional());
       var senders = new ArrayList<Sender<String, byte[]>>(3);
       while (senders.size() < 3) {
         senders.add(
@@ -77,7 +89,7 @@ public class ProducerTest extends RequireBrokerCluster {
                 .timestamp(timestamp)
                 .headers(List.of(header)));
       }
-      producer.transaction(senders);
+      producer.send(senders);
     }
 
     try (var consumer =
@@ -92,4 +104,71 @@ public class ProducerTest extends RequireBrokerCluster {
       Assertions.assertEquals(3, records.size());
     }
   }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  void testInvalidSender() {
+    try (var producer = Producer.builder().brokers(bootstrapServers()).buildTransactional()) {
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> producer.send(List.of((Sender<byte[], byte[]>) Mockito.mock(Sender.class))));
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("offerProducers")
+  void testSingleSend(Producer<byte[], byte[]> producer)
+      throws ExecutionException, InterruptedException {
+    var topic = Utils.randomString(10);
+
+    producer.sender().topic(topic).value(new byte[10]).run().toCompletableFuture().get();
+
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .fromBeginning()
+            .topics(Set.of(topic))
+            .isolation(
+                producer.transactional() ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED)
+            .build()) {
+      Assertions.assertEquals(1, consumer.poll(Duration.ofSeconds(10)).size());
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("offerProducers")
+  void testMultiplesSend(Producer<byte[], byte[]> producer) throws InterruptedException {
+    var topic = Utils.randomString(10);
+    var count = 10;
+    var latch = new CountDownLatch(count);
+    producer
+        .send(
+            IntStream.range(0, count)
+                .mapToObj(i -> producer.sender().topic(topic).value(new byte[10]))
+                .collect(Collectors.toUnmodifiableList()))
+        .forEach(f -> f.whenComplete((m, e) -> latch.countDown()));
+
+    latch.await();
+
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .fromBeginning()
+            .topics(Set.of(topic))
+            .isolation(
+                producer.transactional() ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED)
+            .build()) {
+      Assertions.assertEquals(count, consumer.poll(count, Duration.ofSeconds(10)).size());
+    }
+  }
+
+  private static Stream<Arguments> offerProducers() {
+    return Stream.of(
+        Arguments.of(
+            Named.of("normal producer", Producer.builder().brokers(bootstrapServers()).build())),
+        Arguments.of(
+            Named.of(
+                "transactional producer",
+                Producer.builder().brokers(bootstrapServers()).buildTransactional())));
+  }
 }