X-Git-Url: https://juplo.de/gitweb/?p=demos%2Fkafka%2Fdemos-kafka-payment-system-transfer;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fpayment%2Ftransfer%2FTransferServiceApplication.java;h=e94f703d34a64de2dbe65b6c86ff148c84620b5d;hp=53ff0f4b6673261e7a4b6050f740aa127a483e2a;hb=HEAD;hpb=cbfe4b796266ff7b9689fb69c5a8efee8ebb130a 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 53ff0f4..e94f703 100644 --- 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,17 @@ 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.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; @@ -21,7 +23,14 @@ 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; @@ -31,16 +40,31 @@ import java.util.Properties; @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 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.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); } @@ -50,10 +74,13 @@ public class TransferServiceApplication { 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); @@ -66,6 +93,9 @@ public class TransferServiceApplication TransferConsumer transferConsumer( TransferServiceProperties properties, KafkaConsumer consumer, + AdminClient adminClient, + TransferRepository repository, + LocalStateStoreSettings localStateStoreSettings, ObjectMapper mapper, TransferService productionTransferService, TransferService restoreTransferService) @@ -73,13 +103,19 @@ public class TransferServiceApplication return new TransferConsumer( properties.getTopic(), + properties.getNumPartitions(), + properties.getInstanceIdUriMapping(), consumer, + adminClient, + repository, + Clock.systemDefaultZone(), + localStateStoreSettings.interval, mapper, new TransferConsumer.ConsumerUseCases() { @Override - public void create(Long id, Long payer, Long payee, Integer amount) + public TransferStateChangedEvent create(Long id, Long payer, Long payee, Integer amount) { - productionTransferService.create(id, payer, payee, amount); + return productionTransferService.create(id, payer, payee, amount); } @Override @@ -89,16 +125,17 @@ public class TransferServiceApplication } @Override - public void handleStateChange(Long id, Transfer.State state) + public TransferStateChangedEvent handleStateChange( + TransferStateChangedEvent stateChangedEvent) { - productionTransferService.handleStateChange(id, state); + return productionTransferService.handleStateChange(stateChangedEvent); } }, new TransferConsumer.ConsumerUseCases() { @Override - public void create(Long id, Long payer, Long payee, Integer amount) + public TransferStateChangedEvent create(Long id, Long payer, Long payee, Integer amount) { - restoreTransferService.create(id, payer, payee, amount); + return restoreTransferService.create(id, payer, payee, amount); } @Override @@ -108,9 +145,10 @@ public class TransferServiceApplication } @Override - public void handleStateChange(Long id, Transfer.State state) + public TransferStateChangedEvent handleStateChange( + TransferStateChangedEvent stateChangedEvent) { - restoreTransferService.handleStateChange(id, state); + return restoreTransferService.handleStateChange(stateChangedEvent); } }); } @@ -124,6 +162,60 @@ public class TransferServiceApplication return new KafkaMessagingService(producer, mapper, properties.getTopic()); } + @RequiredArgsConstructor + static class LocalStateStoreSettings + { + final Optional 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(localStateStoreSettings.file, properties.getNumPartitions(), mapper); + } + @Bean TransferService productionTransferService( TransferRepository repository, @@ -143,9 +235,14 @@ 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, + WebClient.create()); }