Switched from single-node (assign) to multi-instance (subscribe)

[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 259b62d..a82f8b1 100644 (file)
--- a/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java
+++ b/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java
@@ -2,15 +2,16 @@ package de.juplo.kafka.payment.transfer;
 
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import de.juplo.kafka.payment.transfer.adapter.KafkaMessagingService;
-import de.juplo.kafka.payment.transfer.adapter.NoOpMessageService;
-import de.juplo.kafka.payment.transfer.adapter.TransferConsumer;
-import de.juplo.kafka.payment.transfer.adapter.TransferController;
+import de.juplo.kafka.payment.transfer.adapter.*;
 import de.juplo.kafka.payment.transfer.domain.Transfer;
-import de.juplo.kafka.payment.transfer.domain.TransferService;
+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.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;
@@ -20,11 +21,10 @@ 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 java.util.Optional;
 import java.util.Properties;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 
 @SpringBootApplication
@@ -32,13 +32,31 @@ import java.util.concurrent.Executors;
 @Slf4j
 public class TransferServiceApplication
 {
+  @Bean(destroyMethod = "close")
+  AdminClient adminClient(TransferServiceProperties properties)
+  {
+    Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
+
+    Properties props = new Properties();
+    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
+
+    return AdminClient.create(props);
+  }
+
   @Bean(destroyMethod = "close")
   KafkaProducer<String, String> producer(TransferServiceProperties properties)
   {
+    Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
+    Assert.hasText(properties.getTopic(), "juplo.transfer.topic must be set");
+    Assert.notNull(properties.getNumPartitions(), "juplo.transfer.num-partitions must be set");
+
     Properties props = new Properties();
-    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, TransferPartitioner.class);
+    props.put(TransferPartitioner.TOPIC, properties.getTopic());
+    props.put(TransferPartitioner.NUM_PARTITIONS, properties.getNumPartitions());
 
     return new KafkaProducer<>(props);
   }
@@ -46,9 +64,15 @@ public class TransferServiceApplication
   @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.bootstrapServers);
-    props.put(ConsumerConfig.GROUP_ID_CONFIG, properties.groupId);
+    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);
@@ -57,32 +81,30 @@ public class TransferServiceApplication
     return new KafkaConsumer<>(props);
   }
 
-  @Bean(destroyMethod = "shutdown")
-  ExecutorService executorService()
-  {
-    return Executors.newFixedThreadPool(1);
-  }
-
   @Bean(destroyMethod = "shutdown")
   TransferConsumer transferConsumer(
       TransferServiceProperties properties,
       KafkaConsumer<String, String> consumer,
-      ExecutorService executorService,
+      AdminClient adminClient,
+      TransferRepository repository,
       ObjectMapper mapper,
       TransferService productionTransferService,
       TransferService restoreTransferService)
   {
     return
         new TransferConsumer(
-            properties.topic,
+            properties.getTopic(),
+            properties.getNumPartitions(),
+            properties.getInstanceIdUriMapping(),
             consumer,
-            executorService,
+            adminClient,
+            repository,
             mapper,
             new TransferConsumer.ConsumerUseCases() {
               @Override
-              public void create(Transfer transfer)
+              public void create(Long id, Long payer, Long payee, Integer amount)
               {
-                productionTransferService.create(transfer);
+                productionTransferService.create(id, payer, payee, amount);
               }
 
               @Override
@@ -92,16 +114,16 @@ public class TransferServiceApplication
               }
 
               @Override
-              public void handle(Transfer transfer)
+              public void handleStateChange(Long id, Transfer.State state)
               {
-                productionTransferService.handle(transfer);
+                productionTransferService.handleStateChange(id, state);
               }
             },
             new TransferConsumer.ConsumerUseCases() {
               @Override
-              public void create(Transfer transfer)
+              public void create(Long id, Long payer, Long payee, Integer amount)
               {
-                restoreTransferService.create(transfer);
+                restoreTransferService.create(id, payer, payee, amount);
               }
 
               @Override
@@ -111,9 +133,9 @@ public class TransferServiceApplication
               }
 
               @Override
-              public void handle(Transfer transfer)
+              public void handleStateChange(Long id, Transfer.State state)
               {
-                restoreTransferService.handle(transfer);
+                restoreTransferService.handleStateChange(id, state);
               }
             });
   }
@@ -124,7 +146,15 @@ public class TransferServiceApplication
       ObjectMapper mapper,
       TransferServiceProperties properties)
   {
-    return new KafkaMessagingService(producer, mapper, properties.topic);
+    return new KafkaMessagingService(producer, mapper, properties.getTopic());
+  }
+
+  @Bean
+  InMemoryTransferRepository inMemoryTransferRepository(
+      TransferServiceProperties properties,
+      ObjectMapper mapper)
+  {
+    return new InMemoryTransferRepository(properties.getNumPartitions(), mapper);
   }
 
   @Bean
@@ -146,9 +176,10 @@ public class TransferServiceApplication
   @Bean
   TransferController transferController(
       TransferService productionTransferService,
-      KafkaMessagingService kafkaMessagingService)
+      KafkaMessagingService kafkaMessagingService,
+      TransferConsumer transferConsumer)
   {
-    return new TransferController(productionTransferService, kafkaMessagingService);
+    return new TransferController(productionTransferService, kafkaMessagingService, transferConsumer);
   }