WIP: Added transactional support to configuration.

README.md

@@ -100,6 +100,7 @@ pulsar.max-backoff-interval-sec=10
 pulsar.consumer-name-delimiter=
 pulsar.namespace=default
 pulsar.tenant=public
+pulsar.transactionCoordinatorEnabled=false
 
 #Consumer
 pulsar.consumer.default.dead-letter-policy-max-redeliver-count=-1

src/main/java/io/github/majusko/pulsar/PulsarAutoConfiguration.java

@@ -36,6 +36,7 @@ public PulsarClient pulsarClient() throws PulsarClientException {
             .operationTimeout(pulsarProperties.getOperationTimeoutSec(), TimeUnit.SECONDS)
             .startingBackoffInterval(pulsarProperties.getStartingBackoffIntervalMs(), TimeUnit.MILLISECONDS)
             .maxBackoffInterval(pulsarProperties.getMaxBackoffIntervalSec(), TimeUnit.SECONDS)
+            .enableTransaction(pulsarProperties.getTransactionCoordinatorEnabled())
             .build();
     }
 }

src/main/java/io/github/majusko/pulsar/properties/PulsarProperties.java

@@ -16,6 +16,7 @@ public class PulsarProperties {
     private String consumerNameDelimiter = "";
     private String namespace = "default";
     private String tenant = "public";
+    private Boolean transactionCoordinatorEnabled = false;
 
     public String getServiceUrl() {
         return serviceUrl;
@@ -112,4 +113,12 @@ public String getTenant() {
     public void setTenant(String tenant) {
         this.tenant = tenant;
     }
+
+    public Boolean getTransactionCoordinatorEnabled() {
+        return transactionCoordinatorEnabled;
+    }
+
+    public void setTransactionCoordinatorEnabled(Boolean transactionCoordinatorEnabled) {
+        this.transactionCoordinatorEnabled = transactionCoordinatorEnabled;
+    }
 }

src/main/resources/application-test.properties

@@ -1,5 +1,7 @@
 pulsar.consumer.default.ack-timeout-ms=1000
 pulsar.consumer.default.dead-letter-policy-max-redeliver-count=1
 
+pulsar.transactionCoordinatorEnabled=true
+
 #Custom Topics
 my.custom.topic.name=custom-topic-name-from-app-prop

src/test/java/io/github/majusko/pulsar/TestConsumers.java

@@ -29,7 +29,7 @@ public class TestConsumers {
     public AtomicInteger failTwiceRetryCount = new AtomicInteger(0);
     public AtomicInteger topicOverflowDueToExceptionRetryCount = new AtomicInteger(0);
 
-    @PulsarConsumer(topic = "topic-one", clazz = MyMsg.class, serialization = Serialization.JSON)
+    @PulsarConsumer(topic = "topic-one", clazz = MyMsg.class, subscriptionType = SubscriptionType.Shared)
     public void topicOneListener(MyMsg myMsg) {
         Assertions.assertNotNull(myMsg);
         mockTopicListenerReceived.set(true);