Kafka PoC

pom.xml

@@ -26,6 +26,7 @@
     <mod-configuration-client.version>5.3.0</mod-configuration-client.version>
     <sonar.exclusions>**/ParallelFileChunkingProcessor.java</sonar.exclusions>
     <sonar.exclusions>**/ModTenantAPI.java</sonar.exclusions>
+    <lombok.version>1.18.12</lombok.version>
   </properties>
 
   <dependencies>
@@ -81,7 +82,53 @@
       <artifactId>dom4j</artifactId>
       <version>2.1.3</version>
     </dependency>
-
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <version>${lombok.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-kafka-client</artifactId>
+      <version>3.8.4</version>
+      <exclusions>
+        <exclusion>
+          <artifactId>org.slf4j</artifactId>
+          <groupId>slf4j-api</groupId>
+        </exclusion>
+        <exclusion>
+          <artifactId>org.slf4j</artifactId>
+          <groupId>slf4j-log4j12</groupId>
+        </exclusion>
+        <exclusion>
+          <artifactId>log4j</artifactId>
+          <groupId>log4j</groupId>
+        </exclusion>
+        <exclusion>
+          <groupId>net.sf.jopt-simple</groupId>
+          <artifactId>jopt-simple</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.zookeeper</groupId>
+          <artifactId>zookeeper</artifactId>
+        </exclusion>
+        <exclusion>
+          <artifactId>jackson-module-scala_2.12</artifactId>
+          <groupId>com.fasterxml.jackson.module</groupId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.folio</groupId>
+      <artifactId>mod-pubsub-client</artifactId>
+      <version>1.2.0</version>
+    </dependency>
+    <dependency>
+      <groupId>org.folio</groupId>
+      <artifactId>data-import-processing-core</artifactId>
+      <version>2.1.5</version>
+    </dependency>
 
     <!-- test dependencies -->
     <dependency>
@@ -190,6 +237,7 @@
           <useIncrementalCompilation>false</useIncrementalCompilation>
           <annotationProcessors>
             <annotationProcessor>io.vertx.codegen.CodeGenProcessor</annotationProcessor>
+            <annotationProcessor>lombok.launch.AnnotationProcessorHider$AnnotationProcessor</annotationProcessor>
           </annotationProcessors>
           <generatedSourcesDirectory>${project.build.directory}/generated-sources/service/processing</generatedSourcesDirectory>
         </configuration>

src/main/java/org/folio/config/ApplicationConfig.java

@@ -1,5 +1,10 @@
 package org.folio.config;
 
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
+import org.folio.kafka.KafkaConfig;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 
@@ -12,4 +17,32 @@
   "org.folio.service.upload",
   "org.folio.service.cleanup"})
 public class ApplicationConfig {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ApplicationConfig.class);
+
+  //TODO: rewrite it in a correct way
+  @Value("${FOLIO_KAFKA_HOST:kafka}")
+  private String kafkaHost;
+  @Value("${FOLIO_KAFKA_PORT:9092}")
+  private String kafkaPort;
+  @Value("${OKAPI_URL:http://okapi:9130}")
+  private String okapiUrl;
+  @Value("${FOLIO_KAFKA_REPLICATION_FACTOR:1}")
+  private int replicationFactor;
+  @Value("${FOLIO_KAFKA_ENV:folio}")
+  private String envId;
+
+  @Bean(name = "newKafkaConfig")
+  public KafkaConfig kafkaConfigBean() {
+    KafkaConfig kafkaConfig = KafkaConfig.builder()
+      .envId(envId)
+      .kafkaHost(kafkaHost)
+      .kafkaPort(kafkaPort)
+      .okapiUrl(okapiUrl)
+      .replicationFactor(replicationFactor)
+      .build();
+    LOGGER.debug("kafkaConfig: " + kafkaConfig);
+
+    return kafkaConfig;
+  }
+
 }

src/main/java/org/folio/kafka/AsyncRecordHandler.java

@@ -0,0 +1,8 @@
+package org.folio.kafka;
+
+import io.vertx.core.Future;
+import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+
+public interface AsyncRecordHandler<K, V> {
+  Future<K> handle(KafkaConsumerRecord<K, V> record);
+}

src/main/java/org/folio/kafka/GlobalLoadSensor.java

@@ -0,0 +1,26 @@
+package org.folio.kafka;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * This is the simplest solution to track the load
+ * Further it makes sense to consider usage of SPI metrics engine profided by Vert.x
+ */
+
+
+public class GlobalLoadSensor {
+  private final AtomicInteger index = new AtomicInteger();
+
+  public int increment() {
+    return index.incrementAndGet();
+  }
+
+  public int decrement() {
+    return index.decrementAndGet();
+  }
+
+  public int current() {
+    return index.get();
+  }
+
+}

src/main/java/org/folio/kafka/KafkaConfig.java

@@ -0,0 +1,61 @@
+package org.folio.kafka;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.ToString;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Getter
+@Builder
+@ToString
+public class KafkaConfig {
+  public static final String KAFKA_CONSUMER_AUTO_OFFSET_RESET = "kafka.consumer.auto.offset.reset";
+  public static final String KAFKA_CONSUMER_AUTO_OFFSET_RESET_DEFAULT = "earliest";
+
+  public static final String KAFKA_CONSUMER_METADATA_MAX_AGE = "kafka.consumer.metadata.max.age.ms";
+  public static final String KAFKA_CONSUMER_METADATA_MAX_AGE_DEFAULT = "30000";
+
+//  public static final String KAFKA_NUMBER_OF_PARTITIONS = "kafka.number.of.patitions";
+  public static final String KAFKA_NUMBER_OF_PARTITIONS = "NUMBER_OF_PARTITIONS";
+  public static final String KAFKA_NUMBER_OF_PARTITIONS_DEFAULT = "10";
+
+  private final String kafkaHost;
+  private final String kafkaPort;
+  private final String okapiUrl;
+  private final int replicationFactor;
+  private final String envId;
+
+  public Map<String, String> getProducerProps() {
+    Map<String, String> producerProps = new HashMap<>();
+    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaUrl());
+    producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+    producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+    producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+    return producerProps;
+  }
+
+  public Map<String, String> getConsumerProps() {
+    Map<String, String> consumerProps = new HashMap<>();
+    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaUrl());
+    consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+    consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, SimpleConfigurationReader.getValue(KAFKA_CONSUMER_AUTO_OFFSET_RESET, KAFKA_CONSUMER_AUTO_OFFSET_RESET_DEFAULT));
+    consumerProps.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, SimpleConfigurationReader.getValue(KAFKA_CONSUMER_METADATA_MAX_AGE, KAFKA_CONSUMER_METADATA_MAX_AGE_DEFAULT));
+    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+    return consumerProps;
+  }
+
+  public String getKafkaUrl() {
+    return kafkaHost + ":" + kafkaPort;
+  }
+
+  public int getNumberOfPartitions() {
+    return Integer.parseInt(SimpleConfigurationReader.getValue(KAFKA_NUMBER_OF_PARTITIONS, KAFKA_NUMBER_OF_PARTITIONS_DEFAULT));
+  }
+
+
+}