Epic / Daway

.github/workflows/build.yml

@@ -1,10 +1,10 @@
-name: Maven Build Artifact
+name: Build Maven Artifact
 
 on:
   pull_request:
     branches:
-      - '*'
+      - '**'
 
 jobs:
   build:
-    uses: valitydev/java-workflow/.github/workflows/maven-service-build.yml@v1
+    uses: valitydev/base-workflow/.github/workflows/maven-service-build.yml@v2

.github/workflows/deploy.yml

@@ -1,18 +1,15 @@
-name: Maven Deploy Artifact
+name: Deploy Docker Image
 
 on:
   push:
     branches:
       - 'master'
       - 'main'
-
-env:
-  REGISTRY: ghcr.io
-  IMAGE_NAME: ${{ github.repository }}
+      - 'epic/**'
 
 jobs:
-  deploy:
-    uses: valitydev/java-workflow/.github/workflows/maven-service-deploy.yml@v1
+  build-and-deploy:
+    uses: valitydev/base-workflow/.github/workflows/maven-service-deploy.yml@v2
     secrets:
       github-token: ${{ secrets.GITHUB_TOKEN }}
       mm-webhook-url: ${{ secrets.MATTERMOST_WEBHOOK_URL }}

pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>dev.vality</groupId>
         <artifactId>service-parent-pom</artifactId>
-        <version>1.0.16</version>
+        <version>2.0.0-BETA-11</version>
     </parent>
 
     <artifactId>newway</artifactId>
@@ -25,7 +25,7 @@
         <db.user>postgres</db.user>
         <db.password>postgres</db.password>
         <db.name>newway</db.name>
-        <db.schema>nw</db.schema>
+        <db.schema>dw</db.schema>
         <local.pg.url>jdbc:postgresql://localhost:5432/newway</local.pg.url>
         <local.pg.port>5432</local.pg.port>
         <checkstyle.config.suppressions.path>./src/main/resources/checkstyle/checkstyle-suppressions.xml
@@ -65,7 +65,7 @@
         <dependency>
             <groupId>de.codecentric</groupId>
             <artifactId>spring-boot-admin-starter-client</artifactId>
-            <version>2.6.6</version>
+            <version>2.7.1</version>
         </dependency>
         <dependency>
             <groupId>org.postgresql</groupId>
@@ -123,6 +123,10 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-prometheus</artifactId>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.msk</groupId>
+            <artifactId>aws-msk-iam-auth</artifactId>
+        </dependency>
 
         <!--dev.vality-->
         <dependency>
@@ -155,17 +159,29 @@
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>damsel</artifactId>
+            <!-- TODO: bump version in parent -->
+            <version>1.597-bfedcb9</version>
         </dependency>
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>fistful-proto</artifactId>
-            <version>1.145-c45166d</version>
+            <version>1.159-936ed9a</version>
         </dependency>
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>xrates-proto</artifactId>
             <version>1.23-bf0d62d</version>
         </dependency>
+        <dependency>
+            <groupId>dev.vality</groupId>
+            <artifactId>limiter-proto</artifactId>
+            <version>1.33-31de59b</version>
+        </dependency>
+        <dependency>
+            <groupId>dev.vality</groupId>
+            <artifactId>exrates-proto</artifactId>
+            <version>1.3-875328b</version>
+        </dependency>
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>shared-resources</artifactId>
@@ -188,7 +204,13 @@
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>testcontainers-annotations</artifactId>
-            <version>1.4.0</version>
+            <version>1.4.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>4.2.0</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

src/main/java/dev/vality/newway/config/ApplicationConfig.java

@@ -1,7 +1,7 @@
 package dev.vality.newway.config;
 
 import dev.vality.damsel.domain_config.RepositorySrv;
-import dev.vality.newway.domain.Nw;
+import dev.vality.newway.domain.Dw;
 import dev.vality.woody.thrift.impl.http.THSpawnClientBuilder;
 import org.jooq.Schema;
 import org.springframework.beans.factory.annotation.Value;
@@ -25,6 +25,6 @@ public RepositorySrv.Iface dominantClient(@Value("${dmt.url}") Resource resource
 
     @Bean
     public Schema schema() {
-        return Nw.NW;
+        return Dw.DW;
     }
 }

src/main/java/dev/vality/newway/config/CacheConfig.java

@@ -2,23 +2,23 @@
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
-import dev.vality.newway.model.InvoiceWrapper;
-import dev.vality.newway.model.InvoicingKey;
-import dev.vality.newway.model.PaymentWrapper;
+import dev.vality.newway.model.PartyShop;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.time.Duration;
+
 @Configuration
 public class CacheConfig {
 
     @Bean
-    public Cache<InvoicingKey, InvoiceWrapper> invoiceDataCache(@Value("${cache.invoice.size}") int cacheSize) {
-        return Caffeine.newBuilder().maximumSize(cacheSize).build();
+    public Cache<String, PartyShop> partyShopDataCache(@Value("${cache.party-shop.size}") int cacheSize,
+                                                       @Value("${cache.party-shop.expire.after.sec}") long expireAfter) {
+        return Caffeine.newBuilder()
+                .maximumSize(cacheSize)
+                .expireAfterWrite(Duration.ofSeconds(expireAfter))
+                .build();
     }
 
-    @Bean
-    public Cache<InvoicingKey, PaymentWrapper> paymentDataCache(@Value("${cache.payment.size}") int cacheSize) {
-        return Caffeine.newBuilder().maximumSize(cacheSize).build();
-    }
 }

src/main/java/dev/vality/newway/config/KafkaConfig.java

@@ -1,20 +1,19 @@
 package dev.vality.newway.config;
 
+import dev.vality.exrates.events.CurrencyEvent;
 import dev.vality.kafka.common.util.ExponentialBackOffDefaultErrorHandlerFactory;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import dev.vality.newway.config.properties.KafkaConsumerProperties;
-import dev.vality.newway.config.properties.KafkaSslProperties;
+import dev.vality.newway.serde.CurrencyExchangeRateEventDeserializer;
 import dev.vality.newway.serde.PayoutEventDeserializer;
 import dev.vality.newway.serde.SinkEventDeserializer;
+import dev.vality.newway.service.FileService;
 import dev.vality.payout.manager.Event;
 import lombok.RequiredArgsConstructor;
-import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.common.config.SslConfigs;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
@@ -24,64 +23,50 @@
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ContainerProperties;
 
-import java.io.File;
-import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 
 @Configuration
 @RequiredArgsConstructor
-@EnableConfigurationProperties(KafkaSslProperties.class)
 @SuppressWarnings("LineLength")
 public class KafkaConfig {
 
+    private final KafkaProperties kafkaProperties;
     private final KafkaConsumerProperties kafkaConsumerProperties;
+    private final FileService fileService;
 
     @Value("${kafka.topics.party-management.consumer.group-id}")
     private String partyConsumerGroup;
-    @Value("${kafka.client-id}")
-    private String clientId;
-    @Value("${kafka.bootstrap-servers}")
-    private String bootstrapServers;
+
+    @Value("${kafka.topics.exrate.consumer.group-id}")
+    private String exrateConsumerGroup;
+
+    @Value("${kafka.topics.withdrawal-adjustment.consumer.group-id}")
+    private String withdrawalAdjustmentConsumerGroup;
+
+    @Value("${kafka.rack.path:/tmp/.kafka_rack_env}")
+    private String rackPath;
 
     @Bean
-    public Map<String, Object> consumerConfigs(KafkaSslProperties kafkaSslProperties) {
-        return createConsumerConfig(kafkaSslProperties);
+    public Map<String, Object> consumerConfigs() {
+        return createConsumerConfig();
     }
 
-    private Map<String, Object> createConsumerConfig(KafkaSslProperties kafkaSslProperties) {
-        Map<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    private Map<String, Object> createConsumerConfig() {
+        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SinkEventDeserializer.class);
         props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroupId());
-        props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
-        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProperties.isEnableAutoCommit());
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProperties.getAutoOffsetReset());
-        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerProperties.getMaxPollRecords());
-        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaConsumerProperties.getSessionTimeoutMs());
-        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerProperties.getMaxPollIntervalMs());
-        configureSsl(props, kafkaSslProperties);
-        return props;
-    }
-
-    private void configureSsl(Map<String, Object> props, KafkaSslProperties kafkaSslProperties) {
-        if (kafkaSslProperties.isEnabled()) {
-            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name());
-            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
-                    new File(kafkaSslProperties.getTrustStoreLocation()).getAbsolutePath());
-            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaSslProperties.getTrustStorePassword());
-            props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, kafkaSslProperties.getKeyStoreType());
-            props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, kafkaSslProperties.getTrustStoreType());
-            props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
-                    new File(kafkaSslProperties.getKeyStoreLocation()).getAbsolutePath());
-            props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, kafkaSslProperties.getKeyStorePassword());
-            props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, kafkaSslProperties.getKeyPassword());
+        String clientRack = fileService.getClientRack(rackPath);
+        if (Objects.nonNull(clientRack)) {
+            props.put(ConsumerConfig.CLIENT_RACK_CONFIG, clientRack);
         }
+        return props;
     }
 
     @Bean
-    public ConsumerFactory<String, MachineEvent> consumerFactory(KafkaSslProperties kafkaSslProperties) {
-        return new DefaultKafkaConsumerFactory<>(consumerConfigs(kafkaSslProperties));
+    public ConsumerFactory<String, MachineEvent> consumerFactory() {
+        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
     }
 
     @Bean
@@ -127,10 +112,23 @@ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,
     }
 
     @Bean
-    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Event>> payoutContainerFactory(
-            KafkaSslProperties kafkaSslProperties) {
+    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, MachineEvent>> withdrawalAdjustmentContainerFactory() {
+        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SinkEventDeserializer.class);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, withdrawalAdjustmentConsumerGroup);
+        String clientRack = fileService.getClientRack(rackPath);
+        if (Objects.nonNull(clientRack)) {
+            props.put(ConsumerConfig.CLIENT_RACK_CONFIG, clientRack);
+        }
+        ConsumerFactory<String, MachineEvent> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
+        return createConcurrentFactory(consumerFactory, kafkaConsumerProperties.getWithdrawalAdjustmentConcurrency());
+    }
+
+    @Bean
+    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Event>> payoutContainerFactory() {
         DefaultKafkaConsumerFactory<String, Event> kafkaConsumerFactory =
-                new DefaultKafkaConsumerFactory<>(createConsumerConfig(kafkaSslProperties));
+                new DefaultKafkaConsumerFactory<>(createConsumerConfig());
         kafkaConsumerFactory.setValueDeserializer(new PayoutEventDeserializer());
         ConcurrentKafkaListenerContainerFactory<String, Event> factory =
                 new ConcurrentKafkaListenerContainerFactory<>();
@@ -157,17 +155,33 @@ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,
     }
 
     @Bean
-    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, MachineEvent>> partyManagementContainerFactory(
-            KafkaSslProperties kafkaSslProperties) {
-        Map<String, Object> configs = createConsumerConfig(kafkaSslProperties);
+    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, MachineEvent>> partyManagementContainerFactory() {
+        Map<String, Object> configs = createConsumerConfig();
         configs.put(ConsumerConfig.GROUP_ID_CONFIG, partyConsumerGroup);
         ConsumerFactory<String, MachineEvent> consumerFactory = new DefaultKafkaConsumerFactory<>(configs);
         return createConcurrentFactory(consumerFactory, kafkaConsumerProperties.getPartyManagementConcurrency());
     }
 
-    private KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, MachineEvent>> createConcurrentFactory(
-            ConsumerFactory<String, MachineEvent> consumerFactory, int threadsNumber) {
-        ConcurrentKafkaListenerContainerFactory<String, MachineEvent> factory =
+    @Bean
+    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, MachineEvent>> limitConfigContainerFactory(
+            ConsumerFactory<String, MachineEvent> consumerFactory) {
+        return createConcurrentFactory(consumerFactory, kafkaConsumerProperties.getLimitConfigConcurrency());
+    }
+
+    @Bean
+    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, CurrencyEvent>> exchangeRateContainerFactory() {
+        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CurrencyExchangeRateEventDeserializer.class);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, exrateConsumerGroup);
+        ConsumerFactory<String, CurrencyEvent> consumerFactory = new DefaultKafkaConsumerFactory<>(props);
+
+        return createConcurrentFactory(consumerFactory, kafkaConsumerProperties.getExrateConcurrency());
+    }
+
+    private <T> KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, T>> createConcurrentFactory(
+            ConsumerFactory<String, T> consumerFactory, int threadsNumber) {
+        ConcurrentKafkaListenerContainerFactory<String, T> factory =
                 new ConcurrentKafkaListenerContainerFactory<>();
         initFactory(consumerFactory, threadsNumber, factory);
         return factory;

src/main/java/dev/vality/newway/config/SchedulerConfig.java

@@ -18,7 +18,7 @@
 @EnableSchedulerLock(defaultLockAtMostFor = "PT5M")
 public class SchedulerConfig {
 
-    public static final String TABLE_NAME = "nw.shedlock";
+    public static final String TABLE_NAME = "dw.shedlock";
 
     @Bean
     public DominantPoller dominantPoller(RepositorySrv.Iface dominantClient,

src/main/java/dev/vality/newway/config/properties/KafkaConsumerProperties.java

@@ -11,12 +11,7 @@
 @ConfigurationProperties(prefix = "kafka.consumer")
 public class KafkaConsumerProperties {
 
-    private String autoOffsetReset;
-    private boolean enableAutoCommit;
     private String groupId;
-    private int maxPollRecords;
-    private int maxPollIntervalMs;
-    private int sessionTimeoutMs;
     private int invoicingConcurrency;
     private int recurrentPaymentToolConcurrency;
     private int partyManagementConcurrency;
@@ -29,5 +24,8 @@ public class KafkaConsumerProperties {
     private int sourceConcurrency;
     private int destinationConcurrency;
     private int withdrawalSessionConcurrency;
+    private int limitConfigConcurrency;
+    private int exrateConcurrency;
+    private int withdrawalAdjustmentConcurrency;
 
 }