1 package de.juplo.kafka.payment.transfer;
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 import org.springframework.web.reactive.function.client.WebClient;
30 import java.io.IOException;
31 import java.nio.file.Files;
32 import java.nio.file.Path;
33 import java.time.Clock;
34 import java.util.Optional;
35 import java.util.Properties;
38 @SpringBootApplication
39 @EnableConfigurationProperties(TransferServiceProperties.class)
41 public class TransferServiceApplication
43 @Bean(destroyMethod = "close")
44 AdminClient adminClient(TransferServiceProperties properties)
46 Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
48 Properties props = new Properties();
49 props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
51 return AdminClient.create(props);
54 @Bean(destroyMethod = "close")
55 KafkaProducer<String, String> producer(TransferServiceProperties properties)
57 Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
58 Assert.hasText(properties.getTopic(), "juplo.transfer.topic must be set");
59 Assert.notNull(properties.getNumPartitions(), "juplo.transfer.num-partitions must be set");
61 Properties props = new Properties();
62 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
63 props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
64 props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
65 props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, TransferPartitioner.class);
66 props.put(TransferPartitioner.TOPIC, properties.getTopic());
67 props.put(TransferPartitioner.NUM_PARTITIONS, properties.getNumPartitions());
69 return new KafkaProducer<>(props);
73 KafkaConsumer<String, String> consumer(TransferServiceProperties properties)
75 Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
76 Assert.hasText(properties.getGroupId(), "juplo.transfer.group-id must be set");
77 Assert.hasText(properties.getGroupInstanceId(), "juplo.transfer.group-instance-id must be set");
79 Properties props = new Properties();
80 props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
81 props.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getGroupId());
82 props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, properties.getGroupInstanceId());
83 props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getCanonicalName());
84 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
85 props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
86 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
87 props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
89 return new KafkaConsumer<>(props);
92 @Bean(destroyMethod = "shutdown")
93 TransferConsumer transferConsumer(
94 TransferServiceProperties properties,
95 KafkaConsumer<String, String> consumer,
96 AdminClient adminClient,
97 TransferRepository repository,
98 LocalStateStoreSettings localStateStoreSettings,
100 TransferService productionTransferService,
101 TransferService restoreTransferService)
104 new TransferConsumer(
105 properties.getTopic(),
106 properties.getNumPartitions(),
107 properties.getInstanceIdUriMapping(),
111 Clock.systemDefaultZone(),
112 localStateStoreSettings.interval,
114 new TransferConsumer.ConsumerUseCases() {
116 public void create(Long id, Long payer, Long payee, Integer amount)
118 productionTransferService.create(id, payer, payee, amount);
122 public Optional<Transfer> get(Long id)
124 return productionTransferService.get(id);
128 public void handleStateChange(Long id, Transfer.State state)
130 productionTransferService.handleStateChange(id, state);
133 new TransferConsumer.ConsumerUseCases() {
135 public void create(Long id, Long payer, Long payee, Integer amount)
137 restoreTransferService.create(id, payer, payee, amount);
141 public Optional<Transfer> get(Long id)
143 return restoreTransferService.get(id);
147 public void handleStateChange(Long id, Transfer.State state)
149 restoreTransferService.handleStateChange(id, state);
155 KafkaMessagingService kafkaMessagingService(
156 KafkaProducer<String, String> producer,
158 TransferServiceProperties properties)
160 return new KafkaMessagingService(producer, mapper, properties.getTopic());
163 @RequiredArgsConstructor
164 static class LocalStateStoreSettings
166 final Optional<File> file;
171 LocalStateStoreSettings localStateStoreSettings(TransferServiceProperties properties)
173 if (properties.getStateStoreInterval() < 1)
175 log.info("juplo.transfer.state-store-interval is < 1: local storage of state is deactivated");
176 return new LocalStateStoreSettings(Optional.empty(), 0);
179 if (!StringUtils.hasText(properties.getLocalStateStorePath()))
181 log.info("juplo.transfer.local-state-store-path is not set: local storage of state is deactivated!");
182 return new LocalStateStoreSettings(Optional.empty(), 0);
185 Path path = Path.of(properties.getLocalStateStorePath());
186 log.info("using {} as local state store", path.toAbsolutePath());
188 if (Files.notExists(path))
192 Files.createFile(path);
194 catch (IOException e)
196 throw new IllegalArgumentException("Could not create local state store: " + path.toAbsolutePath());
200 if (!(Files.isReadable(path) && Files.isWritable(path)))
202 throw new IllegalArgumentException("No R/W-access on local state store: " + path.toAbsolutePath());
205 return new LocalStateStoreSettings(Optional.of(path.toFile()), properties.getStateStoreInterval());
209 InMemoryTransferRepository inMemoryTransferRepository(
210 LocalStateStoreSettings localStateStoreSettings,
211 TransferServiceProperties properties,
214 return new InMemoryTransferRepository(localStateStoreSettings.file, properties.getNumPartitions(), mapper);
218 TransferService productionTransferService(
219 TransferRepository repository,
220 KafkaMessagingService kafkaMessagingService)
222 return new TransferService(repository, kafkaMessagingService);
226 TransferService restoreTransferService(
227 TransferRepository repository,
228 NoOpMessageService noOpMessageService)
230 return new TransferService(repository, noOpMessageService);
234 TransferController transferController(
235 TransferService productionTransferService,
236 KafkaMessagingService kafkaMessagingService,
237 TransferConsumer transferConsumer)
239 return new TransferController(
240 productionTransferService,
241 kafkaMessagingService,
247 public static void main(String[] args)
249 SpringApplication.run(TransferServiceApplication.class, args);