chore(next): Update starters and extract common reusable code to an additional module

Author: juancgalvis

docs/docs/reactive-commons/1-getting-started.md

@@ -120,7 +120,146 @@ If you want to use it, you should read the [Creating a CloudEvent guide](11-crea
 
   </TabItem>
   <TabItem value="kafka" label="Kafka">
-    Comming soon...
+    This quick start tutorial sets up a single node Kafka and runs the sample reactive sender and consumer using Reactive
+Commons.
+
+## Requirements
+
+You need Java JRE installed (Java 17 or later).
+
+## Start Kafka
+
+Start a Kafka broker on your local machine with all the defaults (e.g. port is 9092).
+
+### Containerized
+
+You can run it with Docker or Podman.
+
+The following docker compose has a Kafka broker, a Zookeeper and a Kafka UI.
+
+docker-compose.yml
+```yaml
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.4.1
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "2181:2181"
+
+  kafka:
+    image: confluentinc/cp-kafka:7.4.1
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+    ports:
+      - "9092:9092"
+    depends_on:
+      - zookeeper
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
+      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
+    ports:
+      - "8081:8080"
+    depends_on:
+      - kafka
+```
+
+```shell
+docker-compose up
+```
+
+You may set in /etc/hosts (or equivalent) the following entry:
+
+```txt
+127.0.0.1 kafka
+```
+
+To enter the Kafka UI, open your browser and go to `http://localhost:8081`
+
+## Spring Boot Application
+
+The Spring Boot sample publishes and consumes messages with the `DomainEventBus`. This application illustrates how to
+configure Reactive Commons using RabbitMQ in a Spring Boot environment.
+
+To build your own application using the Reactive Commons API, you need to include a dependency to Reactive Commons.
+
+### Current version
+
+![Maven metadata URL](https://img.shields.io/maven-metadata/v?metadataUrl=https%3A%2F%2Frepo1.maven.org%2Fmaven2%2Forg%2Freactivecommons%2Fasync-commons-rabbit-starter%2Fmaven-metadata.xml)
+
+### Dependency
+
+```groovy
+dependencies {
+    implementation "org.reactivecommons:async-kafka-starter:<version>"
+}
+```
+
+### Configuration properties
+
+Also you need to include the name for your app in the `application.properties`, it is important because this value will
+be used
+to name the application queues inside RabbitMQ:
+
+```properties
+spring.application.name=MyAppName
+```
+
+Or in your `application.yaml`
+
+```yaml
+spring:
+  application:
+    name: MyAppName
+```
+
+You can set the RabbitMQ connection properties through spring boot with
+the [`spring.kafka.*` properties](https://docs.spring.io/spring-boot/docs/current/reference/html/application-properties.html)
+
+```yaml
+spring:
+  kafka:
+    bootstrap-servers: localhost:9092
+```
+
+You can also set it in runtime for example from a secret, so you can create the `KafkaProperties` bean like:
+
+```java title="org.reactivecommons.async.rabbit.config.RabbitProperties"
+
+@Configuration
+public class MyKafkaConfig {
+
+    @Bean
+    @Primary
+    public KafkaProperties myRCKafkaProperties() {
+        KafkaProperties properties = new KafkaProperties();
+        properties.setBootstrapServers(List.of("localhost:9092"));
+        return properties;
+    }
+}
+```
+
+### Multi Broker Instances of Kafka or Multi Domain support
+
+Enables to you the ability to listen events from different domains.
+
+### Cloud Events
+
+Includes the Cloud Events specification.
+
+If you want to use it, you should read the [Creating a CloudEvent guide](11-creating-a-cloud-event.md)
+
   </TabItem>
 </Tabs>
 

docs/docs/reactive-commons/9-configuration-properties.md

@@ -13,7 +13,7 @@ import TabItem from '@theme/TabItem';
 You can customize some predefined variables of Reactive Commons
 
 This can be done by Spring Boot `application.yaml` or by overriding
-the [AsyncProps](https://github.com/reactive-commons/reactive-commons-java/blob/master/async/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/props/AsyncProps.java)
+the [AsyncProps](https://github.com/reactive-commons/reactive-commons-java/blob/master/starters/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/props/AsyncProps.java)
 bean.
 
 ```yaml
@@ -24,7 +24,7 @@ app:
       maxRetries: -1 # -1 will be considered default value. When withDLQRetry is true, it will be retried 10 times. When withDLQRetry is false, it will be retried indefinitely.
       retryDelay: 1000 # interval for message retries, with and without DLQRetry
       listenReplies: true # if you will not use ReqReply patter you can set it to false
-      createTopology: true # if your organization have restricctions with automatic topology creation you can set it to false and create it manually or by your organization process.
+      createTopology: true # if your organization have restrictions with automatic topology creation you can set it to false and create it manually or by your organization process.
       delayedCommands: false # Enable to send a delayed command to an external target
       prefetchCount: 250 # is the maximum number of in flight messages you can reduce it to process less concurrent messages, this settings acts per instance of your service
       flux:
@@ -50,7 +50,7 @@ app:
         password: guest
         virtual-host: /
     # Another domain can be configured with same properties structure that app
-    accounts: # this is a second domain name and can have another independent settup
+    accounts: # this is a second domain name and can have another independent setup
       connectionProperties: # you can override the connection properties of each domain
         host: localhost
         port: 5672
@@ -117,6 +117,80 @@ public AsyncPropsDomain.SecretFiller customFiller() {
 
   </TabItem>
   <TabItem value="kafka" label="Kafka">
-    Comming soon...
+    You can customize some predefined variables of Reactive Commons
+
+This can be done by Spring Boot `application.yaml` or by overriding
+the [AsyncKafkaProps](https://github.com/reactive-commons/reactive-commons-java/blob/master/starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/config/props/AsyncKafkaProps.java)
+bean.
+
+```yaml
+reactive:
+  commons:
+    kafka:
+      app: # this is the name of the default domain
+        withDLQRetry: false # if you want to have dlq queues with retries you can set it to true, you cannot change it after queues are created, because you will get an error, so you should delete topology before the change.
+        maxRetries: -1 # -1 will be considered default value. When withDLQRetry is true, it will be retried 10 times. When withDLQRetry is false, it will be retried indefinitely.
+        retryDelay: 1000 # interval for message retries, with and without DLQRetry
+        checkExistingTopics: true # if you don't want to verify topic existence before send a record you can set it to false
+        createTopology: true # if your organization have restrictions with automatic topology creation you can set it to false and create it manually or by your organization process.
+        domain:
+          ignoreThisListener: false # Allows you to disable event listener for this specific domain
+        connectionProperties: # you can override the connection properties of each domain
+          bootstrap-servers: localhost:9092
+      # Another domain can be configured with same properties structure that app
+      accounts: # this is a second domain name and can have another independent setup
+        connectionProperties: # you can override the connection properties of each domain
+          bootstrap-servers: localhost:9093
+```
+
+You can override this settings programmatically through a `AsyncKafkaPropsDomainProperties` bean.
+
+```java
+package sample;
+
+import org.reactivecommons.async.rabbit.config.RabbitProperties;
+import org.reactivecommons.async.rabbit.config.props.AsyncProps;
+import org.reactivecommons.async.rabbit.config.props.AsyncRabbitPropsDomainProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+
+@Configuration
+public class MyDomainConfig {
+
+    @Bean
+    @Primary
+    public AsyncKafkaPropsDomainProperties customKafkaDomainProperties() {
+        KafkaProperties propertiesApp = new KafkaProperties();
+        propertiesApp.setBootstrapServers(List.of("localhost:9092"));
+
+        RabbitProperties propertiesAccounts = new RabbitProperties();
+        propertiesAccounts.setBootstrapServers(List.of("localhost:9093"));
+
+        return AsyncKafkaPropsDomainProperties.builder()
+                .withDomain("app", AsyncProps.builder()
+                        .connectionProperties(propertiesApp)
+                        .build())
+                .withDomain("accounts", AsyncProps.builder()
+                        .connectionProperties(propertiesAccounts)
+                        .build())
+                .build();
+    }
+}
+```
+
+Additionally, if you want to set only connection properties you can use the `AsyncKafkaPropsDomain.KafkaSecretFiller`
+class.
+
+```java
+
+@Bean
+@Primary
+public AsyncKafkaPropsDomain.KafkaSecretFiller customKafkaFiller() {
+    return (domain, asyncProps) -> {
+        // customize asyncProps here by domain
+    };
+}
+```
+
   </TabItem>
 </Tabs>

starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/config/KafkaProperties.java

@@ -1,6 +1,8 @@
 package org.reactivecommons.async.kafka.config;
 
 
+import org.reactivecommons.async.kafka.config.spring.KafkaPropertiesBase;
+
 public class KafkaProperties extends KafkaPropertiesBase {
 
 }

starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/config/KafkaPropertiesAutoConfig.java

@@ -1,11 +1,12 @@
 package org.reactivecommons.async.kafka.config;
 
 
+import org.reactivecommons.async.kafka.config.spring.KafkaPropertiesBase;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
 @ConfigurationProperties(prefix = "spring.kafka")
 public class KafkaPropertiesAutoConfig extends KafkaPropertiesBase {
     public KafkaPropertiesAutoConfig() {
-        put("bootstrap.servers", "localhost:9092");
+//        put("bootstrap.servers", "localhost:9092");
     }
 }

starters/async-kafka-starter/src/main/java/org/reactivecommons/async/kafka/config/KafkaPropertiesBase.java

@@ -28,6 +28,7 @@
 import org.reactivecommons.async.kafka.converters.json.KafkaJacksonMessageConverter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.ssl.SslBundles;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -42,11 +43,6 @@
 import java.nio.file.Path;
 import java.util.Map;
 
-import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
-import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
-import static org.apache.kafka.clients.producer.ProducerConfig.CLIENT_ID_CONFIG;
-import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
-import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
 import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @Configuration
@@ -57,12 +53,13 @@ public class RCKafkaConfig {
     @Bean
     public ConnectionManager kafkaConnectionManager(AsyncKafkaPropsDomain props,
                                                     MessageConverter converter,
-                                                    KafkaCustomizations customizations) {
+                                                    KafkaCustomizations customizations,
+                                                    SslBundles sslBundles) {
         ConnectionManager connectionManager = new ConnectionManager();
         props.forEach((domain, properties) -> {
-            TopologyCreator creator = createTopologyCreator(properties, customizations);
-            ReactiveMessageSender sender = createMessageSender(properties, converter, creator);
-            ReactiveMessageListener listener = createMessageListener(properties);
+            TopologyCreator creator = createTopologyCreator(properties, customizations, sslBundles);
+            ReactiveMessageSender sender = createMessageSender(properties, converter, creator, sslBundles);
+            ReactiveMessageListener listener = createMessageListener(properties, sslBundles);
             connectionManager.addDomain(domain, listener, sender, creator);
 
             ReactiveMessageSender appDomainSender = connectionManager.getSender(domain);
@@ -98,29 +95,31 @@ public DomainEventBus kafkaDomainEventBus(ConnectionManager manager) {
 
     private static ReactiveMessageSender createMessageSender(AsyncKafkaProps config,
                                                              MessageConverter converter,
-                                                             TopologyCreator topologyCreator) {
+                                                             TopologyCreator topologyCreator,
+                                                             SslBundles sslBundles) {
         KafkaProperties props = config.getConnectionProperties();
-        props.put(CLIENT_ID_CONFIG, config.getAppName());
-        props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        props.put(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
-        SenderOptions<String, byte[]> senderOptions = SenderOptions.create(props);
+        props.setClientId(config.getAppName()); // CLIENT_ID_CONFIG
+        props.getProducer().setKeySerializer(StringSerializer.class); // KEY_SERIALIZER_CLASS_CONFIG;
+        props.getProducer().setValueSerializer(ByteArraySerializer.class); // VALUE_SERIALIZER_CLASS_CONFIG
+        SenderOptions<String, byte[]> senderOptions = SenderOptions.create(props.buildProducerProperties(sslBundles));
         KafkaSender<String, byte[]> kafkaSender = KafkaSender.create(senderOptions);
         return new ReactiveMessageSender(kafkaSender, converter, topologyCreator);
     }
 
     // Receiver
 
-    private static ReactiveMessageListener createMessageListener(AsyncKafkaProps config) {
+    private static ReactiveMessageListener createMessageListener(AsyncKafkaProps config, SslBundles sslBundles) {
         KafkaProperties props = config.getConnectionProperties();
-        props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-        props.put(VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
-        ReceiverOptions<String, byte[]> receiverOptions = ReceiverOptions.create(props);
+        props.getConsumer().setKeyDeserializer(StringDeserializer.class); // KEY_DESERIALIZER_CLASS_CONFIG
+        props.getConsumer().setValueDeserializer(ByteArrayDeserializer.class); // VALUE_DESERIALIZER_CLASS_CONFIG
+        ReceiverOptions<String, byte[]> receiverOptions = ReceiverOptions.create(props.buildConsumerProperties(sslBundles));
         return new ReactiveMessageListener(receiverOptions);
     }
 
     // Shared
-    private static TopologyCreator createTopologyCreator(AsyncKafkaProps config, KafkaCustomizations customizations) {
-        AdminClient adminClient = AdminClient.create(config.getConnectionProperties());
+    private static TopologyCreator createTopologyCreator(AsyncKafkaProps config, KafkaCustomizations customizations,
+                                                         SslBundles sslBundles) {
+        AdminClient adminClient = AdminClient.create(config.getConnectionProperties().buildAdminProperties(sslBundles));
         return new TopologyCreator(adminClient, customizations, config.getCheckExistingTopics());
     }
 
@@ -179,12 +178,13 @@ public static KafkaProperties readPropsFromDotEnv(Path path) throws IOException
         String env = Files.readString(path);
         String[] split = env.split("\n");
         KafkaProperties props = new KafkaProperties();
+        Map<String, String> properties = props.getProperties();
         for (String s : split) {
             if (s.startsWith("#")) {
                 continue;
             }
             String[] split1 = s.split("=", 2);
-            props.put(split1[0], split1[1]);
+            properties.put(split1[0], split1[1]);
         }
         return props;
     }