WIP: instance-mapping from assignor

[demos/kafka/demos-kafka-payment-system-transfer] / src / main / java / de / juplo / kafka / payment / transfer / TransferServiceApplication.java

diff --git a/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java b/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java
index a82f8b1..d7d9f3f 100644 (file)
--- a/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java
+++ b/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java
@@ -7,22 +7,27 @@ import de.juplo.kafka.payment.transfer.domain.Transfer;
 import de.juplo.kafka.payment.transfer.persistence.InMemoryTransferRepository;
 import de.juplo.kafka.payment.transfer.ports.TransferRepository;
 import de.juplo.kafka.payment.transfer.ports.TransferService;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.util.Assert;
-
+import org.springframework.util.StringUtils;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Clock;
 import java.util.Optional;
 import java.util.Properties;
 
@@ -61,44 +66,32 @@ public class TransferServiceApplication
     return new KafkaProducer<>(props);
   }
 
-  @Bean
-  KafkaConsumer<String, String> consumer(TransferServiceProperties properties)
-  {
-    Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
-    Assert.hasText(properties.getGroupId(), "juplo.transfer.group-id must be set");
-    Assert.hasText(properties.getGroupInstanceId(), "juplo.transfer.group-instance-id must be set");
-
-    Properties props = new Properties();
-    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
-    props.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getGroupId());
-    props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, properties.getGroupInstanceId());
-    props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getCanonicalName());
-    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
-    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-
-    return new KafkaConsumer<>(props);
-  }
-
   @Bean(destroyMethod = "shutdown")
   TransferConsumer transferConsumer(
       TransferServiceProperties properties,
-      KafkaConsumer<String, String> consumer,
       AdminClient adminClient,
       TransferRepository repository,
+      LocalStateStoreSettings localStateStoreSettings,
       ObjectMapper mapper,
       TransferService productionTransferService,
       TransferService restoreTransferService)
   {
+    Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
+    Assert.hasText(properties.getGroupId(), "juplo.transfer.group-id must be set");
+    Assert.hasText(properties.getGroupInstanceId(), "juplo.transfer.group-instance-id must be set");
+
     return
         new TransferConsumer(
+            properties.getBootstrapServers(),
+            properties.getGroupId(),
+            properties.getGroupInstanceId(),
             properties.getTopic(),
             properties.getNumPartitions(),
             properties.getInstanceIdUriMapping(),
-            consumer,
             adminClient,
             repository,
+            Clock.systemDefaultZone(),
+            localStateStoreSettings.interval,
             mapper,
             new TransferConsumer.ConsumerUseCases() {
               @Override
@@ -149,12 +142,58 @@ public class TransferServiceApplication
     return new KafkaMessagingService(producer, mapper, properties.getTopic());
   }
 
+  @RequiredArgsConstructor
+  static class LocalStateStoreSettings
+  {
+    final Optional<File> file;
+    final int interval;
+  }
+
+  @Bean
+  LocalStateStoreSettings localStateStoreSettings(TransferServiceProperties properties)
+  {
+    if (properties.getStateStoreInterval() < 1)
+    {
+      log.info("juplo.transfer.state-store-interval is < 1: local storage of state is deactivated");
+      return new LocalStateStoreSettings(Optional.empty(), 0);
+    }
+
+    if (!StringUtils.hasText(properties.getLocalStateStorePath()))
+    {
+      log.info("juplo.transfer.local-state-store-path is not set: local storage of state is deactivated!");
+      return new LocalStateStoreSettings(Optional.empty(), 0);
+    }
+
+    Path path = Path.of(properties.getLocalStateStorePath());
+    log.info("using {} as local state store", path.toAbsolutePath());
+
+    if (Files.notExists(path))
+    {
+      try
+      {
+        Files.createFile(path);
+      }
+      catch (IOException e)
+      {
+        throw new IllegalArgumentException("Could not create local state store: " + path.toAbsolutePath());
+      }
+    }
+
+    if (!(Files.isReadable(path) && Files.isWritable(path)))
+    {
+      throw new IllegalArgumentException("No R/W-access on local state store: " + path.toAbsolutePath());
+    }
+
+    return new LocalStateStoreSettings(Optional.of(path.toFile()), properties.getStateStoreInterval());
+  }
+
   @Bean
   InMemoryTransferRepository inMemoryTransferRepository(
+      LocalStateStoreSettings localStateStoreSettings,
       TransferServiceProperties properties,
       ObjectMapper mapper)
   {
-    return new InMemoryTransferRepository(properties.getNumPartitions(), mapper);
+    return new InMemoryTransferRepository(localStateStoreSettings.file, properties.getNumPartitions(), mapper);
   }
 
   @Bean
@@ -179,7 +218,11 @@ public class TransferServiceApplication
       KafkaMessagingService kafkaMessagingService,
       TransferConsumer transferConsumer)
   {
-    return new TransferController(productionTransferService, kafkaMessagingService, transferConsumer);
+    return new TransferController(
+        productionTransferService,
+        kafkaMessagingService,
+        transferConsumer,
+        WebClient.create());
   }