juplo.de Git - demos/kafka/demos-kafka-payment-system-transfer/blob - src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java

   1 package de.juplo.kafka.payment.transfer;
   2
   3
   4 import com.fasterxml.jackson.databind.ObjectMapper;
   5 import de.juplo.kafka.payment.transfer.adapter.*;
   6 import de.juplo.kafka.payment.transfer.domain.Transfer;
   7 import de.juplo.kafka.payment.transfer.persistence.InMemoryTransferRepository;
   8 import de.juplo.kafka.payment.transfer.ports.TransferRepository;
   9 import de.juplo.kafka.payment.transfer.ports.TransferService;
  10 import lombok.RequiredArgsConstructor;
  11 import lombok.extern.slf4j.Slf4j;
  12 import org.apache.kafka.clients.admin.AdminClient;
  13 import org.apache.kafka.clients.admin.AdminClientConfig;
  14 import org.apache.kafka.clients.consumer.ConsumerConfig;
  15 import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;
  16 import org.apache.kafka.clients.consumer.KafkaConsumer;
  17 import org.apache.kafka.clients.producer.KafkaProducer;
  18 import org.apache.kafka.clients.producer.ProducerConfig;
  19 import org.apache.kafka.common.serialization.StringDeserializer;
  20 import org.apache.kafka.common.serialization.StringSerializer;
  21 import org.springframework.boot.SpringApplication;
  22 import org.springframework.boot.autoconfigure.SpringBootApplication;
  23 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  24 import org.springframework.context.annotation.Bean;
  25 import org.springframework.util.Assert;
  26 import org.springframework.util.StringUtils;
  27
  28 import java.io.File;
  29 import java.io.IOException;
  30 import java.nio.file.Files;
  31 import java.nio.file.Path;
  32 import java.time.Clock;
  33 import java.util.Optional;
  34 import java.util.Properties;
  35
  36
  37 @SpringBootApplication
  38 @EnableConfigurationProperties(TransferServiceProperties.class)
  39 @Slf4j
  40 public class TransferServiceApplication
  41 {
  42   @Bean(destroyMethod = "close")
  43   AdminClient adminClient(TransferServiceProperties properties)
  44   {
  45     Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
  46
  47     Properties props = new Properties();
  48     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
  49
  50     return AdminClient.create(props);
  51   }
  52
  53   @Bean(destroyMethod = "close")
  54   KafkaProducer<String, String> producer(TransferServiceProperties properties)
  55   {
  56     Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
  57     Assert.hasText(properties.getTopic(), "juplo.transfer.topic must be set");
  58     Assert.notNull(properties.getNumPartitions(), "juplo.transfer.num-partitions must be set");
  59
  60     Properties props = new Properties();
  61     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
  62     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  63     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  64     props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, TransferPartitioner.class);
  65     props.put(TransferPartitioner.TOPIC, properties.getTopic());
  66     props.put(TransferPartitioner.NUM_PARTITIONS, properties.getNumPartitions());
  67
  68     return new KafkaProducer<>(props);
  69   }
  70
  71   @Bean
  72   KafkaConsumer<String, String> consumer(TransferServiceProperties properties)
  73   {
  74     Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
  75     Assert.hasText(properties.getGroupId(), "juplo.transfer.group-id must be set");
  76     Assert.hasText(properties.getGroupInstanceId(), "juplo.transfer.group-instance-id must be set");
  77
  78     Properties props = new Properties();
  79     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
  80     props.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getGroupId());
  81     props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, properties.getGroupInstanceId());
  82     props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getCanonicalName());
  83     props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  84     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  85     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  86     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  87
  88     return new KafkaConsumer<>(props);
  89   }
  90
  91   @Bean(destroyMethod = "shutdown")
  92   TransferConsumer transferConsumer(
  93       TransferServiceProperties properties,
  94       KafkaConsumer<String, String> consumer,
  95       AdminClient adminClient,
  96       TransferRepository repository,
  97       LocalStateStoreSettings localStateStoreSettings,
  98       ObjectMapper mapper,
  99       TransferService productionTransferService,
 100       TransferService restoreTransferService)
 101   {
 102     return
 103         new TransferConsumer(
 104             properties.getTopic(),
 105             properties.getNumPartitions(),
 106             properties.getInstanceIdUriMapping(),
 107             consumer,
 108             adminClient,
 109             repository,
 110             Clock.systemDefaultZone(),
 111             localStateStoreSettings.interval,
 112             mapper,
 113             new TransferConsumer.ConsumerUseCases() {
 114               @Override
 115               public void create(Long id, Long payer, Long payee, Integer amount)
 116               {
 117                 productionTransferService.create(id, payer, payee, amount);
 118               }
 119
 120               @Override
 121               public Optional<Transfer> get(Long id)
 122               {
 123                 return productionTransferService.get(id);
 124               }
 125
 126               @Override
 127               public void handleStateChange(Long id, Transfer.State state)
 128               {
 129                 productionTransferService.handleStateChange(id, state);
 130               }
 131             },
 132             new TransferConsumer.ConsumerUseCases() {
 133               @Override
 134               public void create(Long id, Long payer, Long payee, Integer amount)
 135               {
 136                 restoreTransferService.create(id, payer, payee, amount);
 137               }
 138
 139               @Override
 140               public Optional<Transfer> get(Long id)
 141               {
 142                 return restoreTransferService.get(id);
 143               }
 144
 145               @Override
 146               public void handleStateChange(Long id, Transfer.State state)
 147               {
 148                 restoreTransferService.handleStateChange(id, state);
 149               }
 150             });
 151   }
 152
 153   @Bean
 154   KafkaMessagingService kafkaMessagingService(
 155       KafkaProducer<String, String> producer,
 156       ObjectMapper mapper,
 157       TransferServiceProperties properties)
 158   {
 159     return new KafkaMessagingService(producer, mapper, properties.getTopic());
 160   }
 161
 162   @RequiredArgsConstructor
 163   static class LocalStateStoreSettings
 164   {
 165     final Optional<File> file;
 166     final int interval;
 167   }
 168
 169   @Bean
 170   LocalStateStoreSettings localStateStoreSettings(TransferServiceProperties properties)
 171   {
 172     if (properties.getStateStoreInterval() < 1)
 173     {
 174       log.info("juplo.transfer.state-store-interval is < 1: local storage of state is deactivated");
 175       return new LocalStateStoreSettings(Optional.empty(), 0);
 176     }
 177
 178     if (!StringUtils.hasText(properties.getLocalStateStorePath()))
 179     {
 180       log.info("juplo.transfer.local-state-store-path is not set: local storage of state is deactivated!");
 181       return new LocalStateStoreSettings(Optional.empty(), 0);
 182     }
 183
 184     Path path = Path.of(properties.getLocalStateStorePath());
 185     log.info("using {} as local state store", path.toAbsolutePath());
 186
 187     if (Files.notExists(path))
 188     {
 189       try
 190       {
 191         Files.createFile(path);
 192       }
 193       catch (IOException e)
 194       {
 195         throw new IllegalArgumentException("Could not create local state store: " + path.toAbsolutePath());
 196       }
 197     }
 198
 199     if (!(Files.isReadable(path) && Files.isWritable(path)))
 200     {
 201       throw new IllegalArgumentException("No R/W-access on local state store: " + path.toAbsolutePath());
 202     }
 203
 204     return new LocalStateStoreSettings(Optional.of(path.toFile()), properties.getStateStoreInterval());
 205   }
 206
 207   @Bean
 208   InMemoryTransferRepository inMemoryTransferRepository(
 209       LocalStateStoreSettings localStateStoreSettings,
 210       TransferServiceProperties properties,
 211       ObjectMapper mapper)
 212   {
 213     return new InMemoryTransferRepository(localStateStoreSettings.file, properties.getNumPartitions(), mapper);
 214   }
 215
 216   @Bean
 217   TransferService productionTransferService(
 218       TransferRepository repository,
 219       KafkaMessagingService kafkaMessagingService)
 220   {
 221     return new TransferService(repository, kafkaMessagingService);
 222   }
 223
 224   @Bean
 225   TransferService restoreTransferService(
 226       TransferRepository repository,
 227       NoOpMessageService noOpMessageService)
 228   {
 229     return new TransferService(repository, noOpMessageService);
 230   }
 231
 232   @Bean
 233   TransferController transferController(
 234       TransferService productionTransferService,
 235       KafkaMessagingService kafkaMessagingService,
 236       TransferConsumer transferConsumer)
 237   {
 238     return new TransferController(productionTransferService, kafkaMessagingService, transferConsumer);
 239   }
 240
 241
 242   public static void main(String[] args)
 243   {
 244     SpringApplication.run(TransferServiceApplication.class, args);
 245   }
 246 }