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=a82f8b109abeaec2ae9106f4e66f311dcb1347ff;hp=65f683cf35ba12ad7a295856a3a8d6f6c2e1b03f;hb=43ea59755f9673864a3ef95250009f091e99a760;hpb=6191849fee717b080118717c86df79fad12bafc8 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 65f683c..a82f8b1 100644 --- a/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java +++ b/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java @@ -2,19 +2,28 @@ 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.domain.TransferService; -import de.juplo.kafka.payment.transfer.ports.MessagingService; +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.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 java.util.Optional; import java.util.Properties; @@ -23,32 +32,154 @@ 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.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); } @Bean - MessagingService kafkaMessagingService( + KafkaConsumer 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 consumer, + AdminClient adminClient, + TransferRepository repository, + ObjectMapper mapper, + TransferService productionTransferService, + TransferService restoreTransferService) + { + return + new TransferConsumer( + properties.getTopic(), + properties.getNumPartitions(), + properties.getInstanceIdUriMapping(), + consumer, + adminClient, + repository, + mapper, + new TransferConsumer.ConsumerUseCases() { + @Override + public void create(Long id, Long payer, Long payee, Integer amount) + { + productionTransferService.create(id, payer, payee, amount); + } + + @Override + public Optional get(Long id) + { + return productionTransferService.get(id); + } + + @Override + public void handleStateChange(Long id, Transfer.State state) + { + productionTransferService.handleStateChange(id, state); + } + }, + new TransferConsumer.ConsumerUseCases() { + @Override + public void create(Long id, Long payer, Long payee, Integer amount) + { + restoreTransferService.create(id, payer, payee, amount); + } + + @Override + public Optional get(Long id) + { + return restoreTransferService.get(id); + } + + @Override + public void handleStateChange(Long id, Transfer.State state) + { + restoreTransferService.handleStateChange(id, state); + } + }); + } + + @Bean + KafkaMessagingService kafkaMessagingService( KafkaProducer producer, 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 - TransferService transferService( + TransferService productionTransferService( TransferRepository repository, - MessagingService messagingService) + KafkaMessagingService kafkaMessagingService) + { + return new TransferService(repository, kafkaMessagingService); + } + + @Bean + TransferService restoreTransferService( + TransferRepository repository, + NoOpMessageService noOpMessageService) + { + return new TransferService(repository, noOpMessageService); + } + + @Bean + TransferController transferController( + TransferService productionTransferService, + KafkaMessagingService kafkaMessagingService, + TransferConsumer transferConsumer) { - return new TransferService(repository, messagingService); + return new TransferController(productionTransferService, kafkaMessagingService, transferConsumer); }