http://jira.mfd.msk:8088/browse/CLS-758 ControlServiceTest added, testcontainers added to pom.xml

clearing-parent/control-service/pom.xml

@@ -0,0 +1,129 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>control-service</artifactId>
+    <name>control-service</name>
+    <description>Control service</description>
+    <version>SPCEX-3.15.207</version>
+
+    <parent>
+        <artifactId>clearing-parent</artifactId>
+        <groupId>ru.spcex.clearing</groupId>
+        <version>SPCEX-3.15.207</version>
+    </parent>
+
+    <properties>
+        <opencsv.version>5.5.2</opencsv.version>
+        <logstash-logback-encoder.version>7.0.1</logstash-logback-encoder.version>
+        <testcontainers.version>1.20.2</testcontainers.version>
+    </properties>
+
+    <dependencies>
+<!--         Spring boot -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+
+        <!-- Database -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.mchange</groupId>
+            <artifactId>c3p0</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-imdg-api-hazelcast-impl</artifactId>
+        </dependency>
+
+        <!-- CSV -->
+        <dependency>
+            <groupId>com.opencsv</groupId>
+            <artifactId>opencsv</artifactId>
+            <version>${opencsv.version}</version>
+        </dependency>
+
+        <!-- Kafka -->
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-messaging</artifactId>
+        </dependency>
+
+        <!-- special logging -->
+        <dependency>
+            <groupId>net.logstash.logback</groupId>
+            <artifactId>logstash-logback-encoder</artifactId>
+            <version>${logstash-logback-encoder.version}</version>
+        </dependency>
+
+        <!-- Miscellaneous -->
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-enum</artifactId>
+        </dependency>
+
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>ru.spcex.clearing</groupId>
+            <artifactId>test-clearing</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>postgresql</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <excludes>
+                    <exclude>application.properties</exclude>
+                </excludes>
+                <filtering>false</filtering>
+            </resource>
+        </resources>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <finalName>${project.artifactId}</finalName>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/ControlServiceApplication.java

@@ -0,0 +1,11 @@
+package ru.spcex.clearing.control;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class ControlServiceApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(ControlServiceApplication.class, args);
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/DataSourceConfiguration.java

@@ -0,0 +1,48 @@
+package ru.spcex.clearing.control.config;
+
+import com.mchange.v2.c3p0.ComboPooledDataSource;
+import java.beans.PropertyVetoException;
+import javax.sql.DataSource;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+
+@Configuration
+public class DataSourceConfiguration {
+    private final ControlServiceSettings settings;
+
+    public DataSourceConfiguration(ControlServiceSettings settings) {
+        this.settings = settings;
+    }
+
+    @Bean
+    public DataSource getDataSource() {
+        ComboPooledDataSource cpds = new ComboPooledDataSource();
+        try {
+            cpds.setDriverClass("org.postgresql.Driver");
+            cpds.setJdbcUrl(settings.getDatabase().getUrl());
+            cpds.setUser(settings.getDatabase().getUsername());
+            cpds.setPassword(settings.getDatabase().getPassword());
+
+            cpds.setMaxPoolSize(100);
+            cpds.setMinPoolSize(50);
+            cpds.setAcquireIncrement(5);
+        } catch (PropertyVetoException e) {
+            throw new RuntimeException(e);
+        }
+
+        return cpds;
+    }
+
+    @Bean
+    public JdbcTemplate getJdbcTemplate() {
+        return new JdbcTemplate(getDataSource());
+    }
+
+    @Bean
+    public NamedParameterJdbcTemplate getNamedParameterJdbcTemplate() {
+        return new NamedParameterJdbcTemplate(getDataSource());
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/ImdgConfig.java

@@ -0,0 +1,58 @@
+package ru.spcex.clearing.control.config;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
+import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService;
+
+@Configuration
+public class ImdgConfig {
+    @Bean("taskExecutorHazelcastClientInitializer")
+    public ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer() {
+        return createThreadPoolTaskExecutor(1, true);
+    }
+
+    @Bean("taskExecutorIdGeneratorAwaiter")
+    public ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter() {
+        return createThreadPoolTaskExecutor(1, false);
+    }
+
+    @Bean("imdgProvider")
+    public ImdgProvider imdgProvider(@Qualifier("taskExecutorHazelcastClientInitializer") ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer,
+                                     @Qualifier("taskExecutorIdGeneratorAwaiter") ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter,
+                                     ControlServiceSettings settings) {
+        HazelcastClientParams params = new HazelcastClientParams();
+        params.setClusterMembers(settings.getHazelcast().getClusterMembers());
+        params.setLogin(settings.getHazelcast().getLogin());
+        params.setPassword(settings.getHazelcast().getPassword());
+        return new HazelcastService(taskExecutorHazelcastClientInitializer, taskExecutorIdGeneratorAwaiter, params);
+    }
+
+    @Bean("executor")
+    public ThreadPoolTaskExecutor executor(ControlServiceSettings settings) {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setMaxPoolSize(10);
+        executor.setCorePoolSize(10);
+        executor.setThreadNamePrefix("control-service");
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        executor.setAwaitTerminationSeconds(300);
+        executor.initialize();
+        return executor;
+    }
+
+    private static ThreadPoolTaskExecutor createThreadPoolTaskExecutor(int maxPoolSz, boolean waitForCompletion) {
+        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
+        if (maxPoolSz > 2) {
+            pool.setKeepAliveSeconds(60);
+            pool.setAllowCoreThreadTimeOut(true);
+        }
+        pool.setCorePoolSize(maxPoolSz);
+        pool.setWaitForTasksToCompleteOnShutdown(waitForCompletion);
+        return pool;
+    }
+
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaConfig.java

@@ -0,0 +1,30 @@
+package ru.spcex.clearing.control.config;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Scope;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
+
+@Configuration
+public class KafkaConfig {
+    @Autowired
+    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+    @Bean
+    public Consumer<String, Object> createConsumer(ControlServiceSettings settings) {
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
+    }
+
+    @Autowired
+    @Bean
+    public Producer<String, Object> createProducer(ControlServiceSettings settings) {
+        return settings.getKafkaProducer() != null ?
+            KafkaProducerFactory.producer(settings.getKafkaProducer()) :
+            null;
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaSenderConfig.java

@@ -0,0 +1,57 @@
+package ru.spcex.clearing.control.config;
+
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
+import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+import ru.spcex.platform.imdg.api.ImdgId;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+
+@Configuration
+public class KafkaSenderConfig {
+    Logger log = LoggerFactory.getLogger(getClass());
+
+    private final ImdgProvider imdgProvider;
+
+    @Autowired
+    public KafkaSenderConfig(ImdgProvider imdgProvider) {
+        this.imdgProvider = imdgProvider;
+    }
+
+    @Bean
+    public ProducerFactory<String, Object> pf(ControlServiceSettings settings) {
+        if (settings.getKafkaProducer() == null) {
+            return null;
+        }
+        KafkaProducerSettings kafkaSettings = settings.getKafkaProducer();
+        return KafkaProducerFactory.producerFactory(kafkaSettings);
+    }
+
+    @Bean("kafkaTemplate")
+    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> pf) {
+        if (pf == null) {
+            return null;
+        }
+        return new KafkaTemplate<>(pf);
+    }
+
+    @Bean
+    public Supplier<KafkaSender> kafkaSenderSupplier(KafkaTemplate<String, Object> kafkaTemplate,
+                                                     ImdgProvider imdgProvider) {
+        ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
+        return () -> KafkaSender
+            .setup()
+            .setKafkaTemplate(kafkaTemplate)
+            .idGenerator(imdgIdGenerator::nextId)
+            .saveRequestInfo(false)
+            .build();
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/settings/ControlServiceSettings.java

@@ -0,0 +1,65 @@
+package ru.spcex.clearing.control.config.settings;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import org.springframework.stereotype.Component;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
+import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
+
+@Component
+@ConfigurationProperties("control-service")
+public class ControlServiceSettings {
+    @NestedConfigurationProperty
+    private Database database;
+
+    @NestedConfigurationProperty
+    private HazelcastClientParams hazelcast;
+
+    @NestedConfigurationProperty
+    private KafkaConsumerSettings kafkaConsumer;
+    @NestedConfigurationProperty
+    private KafkaProducerSettings kafkaProducer;
+
+    private String resultDirectory;
+
+    public Database getDatabase() {
+        return database;
+    }
+
+    public void setDatabase(Database database) {
+        this.database = database;
+    }
+
+    public HazelcastClientParams getHazelcast() {
+        return hazelcast;
+    }
+
+    public void setHazelcast(HazelcastClientParams hazelcast) {
+        this.hazelcast = hazelcast;
+    }
+
+    public KafkaConsumerSettings getKafkaConsumer() {
+        return kafkaConsumer;
+    }
+
+    public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
+        this.kafkaConsumer = kafkaConsumer;
+    }
+
+    public KafkaProducerSettings getKafkaProducer() {
+        return kafkaProducer;
+    }
+
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
+    }
+
+    public String getResultDirectory() {
+        return resultDirectory;
+    }
+
+    public void setResultDirectory(String resultDirectory) {
+        this.resultDirectory = resultDirectory;
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/settings/Database.java

@@ -0,0 +1,31 @@
+package ru.spcex.clearing.control.config.settings;
+
+public class Database {
+    private String username;
+    private String password;
+    private String url;
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/mapper/ResultContainerMapper.java

@@ -0,0 +1,19 @@
+package ru.spcex.clearing.control.mapper;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.stereotype.Component;
+import ru.spcex.clearing.control.model.ResultContainer;
+
+@Component
+public class ResultContainerMapper implements RowMapper<ResultContainer> {
+    @Override
+    public ResultContainer mapRow(ResultSet rs, int rowNum) throws SQLException {
+        return new ResultContainer(
+            rs.getLong("first_table"),
+            rs.getLong("second_table"),
+            rs.getLong("diff")
+        );
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/model/ResultContainer.java

@@ -0,0 +1,145 @@
+package ru.spcex.clearing.control.model;
+
+import com.opencsv.bean.CsvBindByName;
+import com.opencsv.bean.CsvBindByPosition;
+import java.util.Objects;
+import ru.spcex.platform.enumeration.SessionType;
+
+public class ResultContainer {
+    @CsvBindByName(column = "sessionId")
+    @CsvBindByPosition(position = 0)
+    private Long sessionId;
+
+    @CsvBindByName(column = "sessionType")
+    @CsvBindByPosition(position = 1)
+    private SessionType sessionType;
+
+    @CsvBindByName(column = "firstTableTitle")
+    @CsvBindByPosition(position = 2)
+    private String firstTableTitle;
+
+    @CsvBindByName(column = "firstTableRecordQuantity")
+    @CsvBindByPosition(position = 3)
+    private Long firstTableRecordQuantity;
+
+    @CsvBindByName(column = "secondTableRecordTitle")
+    @CsvBindByPosition(position = 4)
+    private String secondTableRecordTitle;
+
+    @CsvBindByName(column = "secondTableRecordQuantity")
+    @CsvBindByPosition(position = 5)
+    private Long secondTableRecordQuantity;
+
+    @CsvBindByName(column = "quantityDifference")
+    @CsvBindByPosition(position = 6)
+    private Long quantityDifference;
+
+    public ResultContainer(Long sessionId,
+                           SessionType sessionType,
+                           String firstTableTitle,
+                           Long firstTableRecordQuantity,
+                           String secondTableRecordTitle,
+                           Long secondTableRecordQuantity,
+                           Long quantityDifference) {
+        this.sessionId = sessionId;
+        this.sessionType = sessionType;
+        this.firstTableTitle = firstTableTitle;
+        this.firstTableRecordQuantity = firstTableRecordQuantity;
+        this.secondTableRecordTitle = secondTableRecordTitle;
+        this.secondTableRecordQuantity = secondTableRecordQuantity;
+        this.quantityDifference = quantityDifference;
+    }
+
+    public ResultContainer(Long firstTableRecordQuantity,
+                           Long secondTableRecordQuantity,
+                           Long quantityDifference) {
+        this.firstTableRecordQuantity = firstTableRecordQuantity;
+        this.secondTableRecordQuantity = secondTableRecordQuantity;
+        this.quantityDifference = quantityDifference;
+    }
+
+    public ResultContainer() {
+    }
+
+    public Long getSessionId() {
+        return sessionId;
+    }
+
+    public void setSessionId(Long sessionId) {
+        this.sessionId = sessionId;
+    }
+
+    public SessionType getSessionType() {
+        return sessionType;
+    }
+
+    public void setSessionType(SessionType sessionType) {
+        this.sessionType = sessionType;
+    }
+
+    public String getFirstTableTitle() {
+        return firstTableTitle;
+    }
+
+    public void setFirstTableTitle(String firstTableTitle) {
+        this.firstTableTitle = firstTableTitle;
+    }
+
+    public Long getFirstTableRecordQuantity() {
+        return firstTableRecordQuantity;
+    }
+
+    public void setFirstTableRecordQuantity(Long firstTableRecordQuantity) {
+        this.firstTableRecordQuantity = firstTableRecordQuantity;
+    }
+
+    public String getSecondTableRecordTitle() {
+        return secondTableRecordTitle;
+    }
+
+    public void setSecondTableRecordTitle(String secondTableRecordTitle) {
+        this.secondTableRecordTitle = secondTableRecordTitle;
+    }
+
+    public Long getSecondTableRecordQuantity() {
+        return secondTableRecordQuantity;
+    }
+
+    public void setSecondTableRecordQuantity(Long secondTableRecordQuantity) {
+        this.secondTableRecordQuantity = secondTableRecordQuantity;
+    }
+
+    public Long getQuantityDifference() {
+        return quantityDifference;
+    }
+
+    public void setQuantityDifference(Long quantityDifference) {
+        this.quantityDifference = quantityDifference;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ResultContainer that = (ResultContainer) o;
+        return Objects.equals(sessionId, that.sessionId) && sessionType == that.sessionType && Objects.equals(firstTableTitle, that.firstTableTitle) && Objects.equals(firstTableRecordQuantity, that.firstTableRecordQuantity) && Objects.equals(secondTableRecordTitle, that.secondTableRecordTitle) && Objects.equals(secondTableRecordQuantity, that.secondTableRecordQuantity) && Objects.equals(quantityDifference, that.quantityDifference);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(sessionId, sessionType, firstTableTitle, firstTableRecordQuantity, secondTableRecordTitle, secondTableRecordQuantity, quantityDifference);
+    }
+
+    @Override
+    public String toString() {
+        return "ResultContainer{" +
+            "sessionId=" + sessionId +
+            ", sessionType=" + sessionType +
+            ", firstTableTitle='" + firstTableTitle + '\'' +
+            ", firstTableRecordQuantity=" + firstTableRecordQuantity +
+            ", secondTableRecordTitle='" + secondTableRecordTitle + '\'' +
+            ", secondTableRecordQuantity=" + secondTableRecordQuantity +
+            ", quantityDifference=" + quantityDifference +
+            '}';
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/repository/ControlRepository.java

@@ -0,0 +1,115 @@
+package ru.spcex.clearing.control.repository;
+
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+import ru.spcex.clearing.control.mapper.ResultContainerMapper;
+import ru.spcex.clearing.control.model.ResultContainer;
+
+@Repository
+public class ControlRepository {
+    private final JdbcTemplate jdbcTemplate;
+    private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
+    private final ResultContainerMapper resultContainerMapper;
+
+    public ControlRepository(JdbcTemplate jdbcTemplate,
+                             NamedParameterJdbcTemplate namedParameterJdbcTemplate,
+                             ResultContainerMapper resultContainerMapper) {
+        this.jdbcTemplate = jdbcTemplate;
+        this.namedParameterJdbcTemplate = namedParameterJdbcTemplate;
+        this.resultContainerMapper = resultContainerMapper;
+    }
+
+    @Transactional
+    public ResultContainer getFinlCountDiff() {
+        String sqlQuery = """
+            WITH s_trades_count AS (SELECT COUNT(*) FROM s_trades WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
+                 execution_deposit_count AS (SELECT COUNT(*) FROM execution_deposit WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE))
+            SELECT (SELECT * FROM s_trades_count)                                AS first_table,
+                   (SELECT * FROM execution_deposit_count)                       AS second_table,
+                   (SELECT s_trades_count.count - execution_deposit_count.count) AS diff
+            FROM s_trades_count,
+                 execution_deposit_count;
+            """;
+        ResultContainer resultContainer = jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("s_trades");
+            resultContainer.setSecondTableRecordTitle("execution_deposit");
+        }
+
+        return resultContainer;
+    }
+
+    @Transactional
+    public ResultContainer getIpoTrdCountDiff() {
+        String sqlQuery = """
+            WITH s_trades_fond_count AS (SELECT COUNT(*) FROM s_trades WHERE section = 'FOND' AND CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
+                 execution_fond_count AS (SELECT COUNT(*) FROM execution_fond WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE))
+            SELECT (SELECT * FROM s_trades_fond_count)                             AS first_table,
+                   (SELECT * FROM execution_fond_count)                            AS second_table,
+                   (SELECT s_trades_fond_count.count - execution_fond_count.count) AS diff
+            FROM s_trades_fond_count,
+                 execution_fond_count;
+            """;
+        ResultContainer resultContainer = jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("s_trades");
+            resultContainer.setSecondTableRecordTitle("execution_fond");
+        }
+
+        return resultContainer;
+    }
+
+    @Transactional
+    public ResultContainer getCurrInitCountDiff() {
+        String sqlQuery = """
+            WITH s_trades_curr_count AS (SELECT COUNT(*) FROM s_trades WHERE section = 'CURR' AND CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
+                 execution_currency_count AS (SELECT COUNT(*) FROM execution_currency WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE))
+            SELECT (SELECT * FROM s_trades_curr_count)                                 AS first_table,
+                   (SELECT * FROM execution_currency_count)                            AS second_table,
+                   (SELECT s_trades_curr_count.count - execution_currency_count.count) AS diff
+            FROM s_trades_curr_count,
+                 execution_currency_count;
+            """;
+        ResultContainer resultContainer = jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("s_trades");
+            resultContainer.setSecondTableRecordTitle("execution_currency");
+        }
+
+        return resultContainer;
+    }
+
+    @Transactional
+    public ResultContainer getInternalCountDiff(Long sessionId) {
+        return new ResultContainer(1L, 2L, 3L);
+    }
+
+    @Transactional
+    public ResultContainer getFinlInternalCountDiff(Long sessionId) {
+        String sqlQuery = """
+            WITH payment_instruction_ids AS (SELECT id FROM payment_instruction WHERE session_id = :sessionId AND CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
+                 s_df03_ids AS (SELECT s_df03.id
+                                FROM payment_instruction_ids
+                                         LEFT JOIN s_df03 ON payment_instruction_ids.id = s_df03.payment_instruction_id AND CAST(generation_time AS DATE) = CAST(NOW() AS DATE)),
+                 payment_instruction_count AS (SELECT COUNT(*) FROM payment_instruction_ids),
+                 s_df03_count AS (SELECT COUNT(*) FROM s_df03_ids)
+            SELECT (SELECT payment_instruction_count.count FROM payment_instruction_count) AS first_table,
+                   (SELECT s_df03_count.count FROM s_df03_count)                           AS second_table,
+                   (SELECT payment_instruction_count.count - s_df03_count.count)           AS diff
+            FROM payment_instruction_count,
+                 s_df03_count;
+            """;
+        ResultContainer resultContainer = namedParameterJdbcTemplate.queryForObject(sqlQuery,
+            new MapSqlParameterSource("sessionId", sessionId),
+            resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("payment_instruction");
+            resultContainer.setSecondTableRecordTitle("s_df03");
+        }
+
+        return resultContainer;
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/CommandService.java

@@ -0,0 +1,33 @@
+package ru.spcex.clearing.control.service;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.stereotype.Service;
+import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
+import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+import ru.spcex.platform.enumeration.SessionType;
+import ru.spcex.platform.enumeration.Task;
+
+@Service
+public class CommandService extends QueueConsumer implements InitializingBean {
+    private final ControlService controlService;
+
+    public CommandService(Consumer<String, Object> kafkaQueue, ControlService controlService) {
+        super(kafkaQueue);
+        this.controlService = controlService;
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        callback(LauncherCommandRequest.class)
+            .setConsumer(request -> controlService.generalControl(SessionType.valueOf(request.getRequestPayload().getSessionType())))
+            .forDestination(Task.generalControl_GLCL.topic(), callbacks::put);
+        callback(LauncherCommandRequest.class)
+            .setConsumer(request -> controlService.internalControl(request.getRequestPayload().getSessionId()))
+            .forDestination(Task.internalControl_ILCL.topic(), callbacks::put);
+        callback(LauncherCommandRequest.class)
+            .setConsumer(request -> controlService.internalControlFinl(request.getRequestPayload().getSessionId()))
+            .forDestination(Task.internalFinlControl_IFCL.topic(), callbacks::put);
+        init();
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/ControlService.java

@@ -0,0 +1,99 @@
+package ru.spcex.clearing.control.service;
+
+import com.opencsv.bean.StatefulBeanToCsv;
+import com.opencsv.bean.StatefulBeanToCsvBuilder;
+import com.opencsv.exceptions.CsvDataTypeMismatchException;
+import com.opencsv.exceptions.CsvRequiredFieldEmptyException;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.Writer;
+import java.nio.file.Path;
+import java.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+import ru.spcex.clearing.control.model.ResultContainer;
+import ru.spcex.clearing.control.repository.ControlRepository;
+import ru.spcex.clearing.control.util.CustomMappingStrategy;
+import ru.spcex.platform.enumeration.SessionType;
+
+@Service
+public class ControlService {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final ControlRepository controlRepository;
+    private final ControlServiceSettings settings;
+    private final Path resultDirectory;
+
+    public ControlService(ControlRepository controlRepository,
+                          ControlServiceSettings settings) {
+        this.controlRepository = controlRepository;
+        this.settings = settings;
+        resultDirectory = Path.of(settings.getResultDirectory());
+    }
+
+    public File generalControl(SessionType sessionType) {
+        log.info("Starting general control with session type {}", sessionType);
+
+        ResultContainer resultContainer = switch (sessionType) {
+            case FINL -> controlRepository.getFinlCountDiff();
+            case IPOB, IPOT, IPO0, TRDT -> controlRepository.getIpoTrdCountDiff();
+            case CURR -> controlRepository.getCurrInitCountDiff();
+            default -> {
+                log.error("Unknown session type {}", sessionType);
+                throw new IllegalStateException("Unexpected value: " + sessionType);
+            }
+        };
+
+        if (resultContainer.getQuantityDifference() != 0) {
+            resultContainer.setSessionType(sessionType);
+            return writeToCsvFile(resultContainer);
+        }
+
+        return null;
+    }
+
+    public File internalControl(Long sessionId) {
+        log.info("Starting internal control with session id {}", sessionId);
+
+        ResultContainer resultContainer = controlRepository.getInternalCountDiff(sessionId);
+
+        if (resultContainer.getQuantityDifference() != 0) {
+            resultContainer.setSessionId(sessionId);
+            return writeToCsvFile(resultContainer);
+        }
+
+        return null;
+    }
+
+    public File internalControlFinl(Long sessionId) {
+        log.info("Starting internal control FINL with session id {}", sessionId);
+
+        ResultContainer resultContainer = controlRepository.getFinlInternalCountDiff(sessionId);
+
+        if (resultContainer.getQuantityDifference() != 0) {
+            resultContainer.setSessionId(sessionId);
+            return writeToCsvFile(resultContainer);
+        }
+
+        return null;
+    }
+
+    private File writeToCsvFile(ResultContainer resultContainer) {
+        File resultFile = resultDirectory.resolve("result-" + Instant.now().toEpochMilli() + ".csv").toFile();
+        try (Writer writer = new FileWriter(resultFile)) {
+            CustomMappingStrategy<ResultContainer> mappingStrategy = new CustomMappingStrategy<>();
+            mappingStrategy.setType(ResultContainer.class);
+
+            StatefulBeanToCsv<ResultContainer> sbc = new StatefulBeanToCsvBuilder<ResultContainer>(writer)
+                .withMappingStrategy(mappingStrategy)
+                .build();
+            sbc.write(resultContainer);
+        } catch (CsvRequiredFieldEmptyException | CsvDataTypeMismatchException | IOException e) {
+            log.error(e.getMessage(), e);
+        }
+        return resultFile;
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/util/CustomMappingStrategy.java

@@ -0,0 +1,52 @@
+package ru.spcex.clearing.control.util;
+
+import com.opencsv.bean.AbstractCsvConverter;
+import com.opencsv.bean.BeanField;
+import com.opencsv.bean.ColumnPositionMappingStrategy;
+import com.opencsv.bean.ConverterNumber;
+import com.opencsv.bean.CsvBindByName;
+import com.opencsv.bean.CsvBindByPosition;
+import com.opencsv.bean.CsvConverter;
+import com.opencsv.bean.CsvNumber;
+import com.opencsv.exceptions.CsvBadConverterException;
+import com.opencsv.exceptions.CsvChainedException;
+import com.opencsv.exceptions.CsvFieldAssignmentException;
+import com.opencsv.exceptions.CsvRequiredFieldEmptyException;
+import java.lang.reflect.Field;
+import java.math.BigDecimal;
+import java.util.Locale;
+
+public class CustomMappingStrategy<T> extends ColumnPositionMappingStrategy<T> {
+    public String[] generateHeader(T bean) throws CsvRequiredFieldEmptyException {
+        super.generateHeader(bean);
+        String[] headers = new String[this.getFieldMap().values().size()];
+        for (BeanField<T, Integer> field : this.getFieldMap().values()) {
+            CsvBindByPosition positionAnnotation = field.getField().getAnnotation(CsvBindByPosition.class);
+            CsvBindByName nameAnnotation = field.getField().getAnnotation(CsvBindByName.class);
+            headers[positionAnnotation.position()] = nameAnnotation.column();
+        }
+        return headers;
+    }
+
+    @Override
+    public String[] transmuteBean(T bean) throws CsvFieldAssignmentException, CsvChainedException {
+        return super.transmuteBean(bean);
+    }
+
+    @Override
+    protected CsvConverter determineConverter(Field field, Class<?> elementType, String locale, String writeLocale, Class<? extends AbstractCsvConverter> customConverter) throws CsvBadConverterException {
+        CsvConverter csvConverter = super.determineConverter(field, elementType, locale, writeLocale, customConverter);
+        if (csvConverter instanceof ConverterNumber && field.getType() == BigDecimal.class) {
+            CsvNumber csvNumberAnnotation = field.getAnnotation(CsvNumber.class);
+            String formatValue = csvNumberAnnotation.value();
+            String writeFormat = csvNumberAnnotation.writeFormat();
+            csvConverter = new ConverterNumber(BigDecimal.class,
+                Locale.getDefault().toString(),
+                Locale.getDefault().toString(),
+                Locale.getDefault(),
+                formatValue,
+                writeFormat);
+        }
+        return csvConverter;
+    }
+}

clearing-parent/control-service/src/main/resources/application.properties

@@ -0,0 +1,27 @@
+# Result directory
+control-service.result-directory=C:\\Users\\inikolaev\\Desktop\\result
+
+# Database settings
+control-service.database.username=clearing
+control-service.database.password=Aa111111
+control-service.database.url=jdbc:postgresql://localhost:5433/clearing
+
+# Hazelcast settings
+control-service.hazelcast.cluster-members=10.200.200.181:5701
+control-service.hazelcast.login=dev
+control-service.hazelcast.password=dev-pass
+
+# Kafka consumer settings
+control-service.kafka-consumer.bootstrap-servers=localhost:19092
+control-service.kafka-consumer.group-id=dev-group-balance-service
+control-service.kafka-consumer.enable-auto-commit=false
+control-service.kafka-consumer.session-timeout-ms=30000
+control-service.kafka-consumer.auto-offset-reset=latest
+
+# Kafka producer settings
+control-service.kafka-producer.bootstrap-servers=localhost:19092
+control-service.kafka-producer.acks=all
+control-service.kafka-producer.retries=0
+control-service.kafka-producer.batch-size=16384
+control-service.kafka-producer.linger-ms=1
+control-service.kafka-producer.buffer-memory=33554432

clearing-parent/control-service/src/main/resources/logback.xml

@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <property name="LOG_PATH" value="./log" />
+    <property name="FILE_NAME" value="control-service" />
+    <property name="CONSOLE_LOG_PATTERN" value="%date{HH:mm:ss.SSS} [%thread] %-5level %class{0}:%line - %message%n" />
+    <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %class{0}:%msg%n" />
+    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <!--            |%X{ru.nbch.scoring.web.logging.mdc_key}-->
+            <Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
+            <charset>utf-8</charset>
+        </encoder>
+    </appender>
+    <!-- first FILE TEXT appender -->
+    <appender name="TEXT_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>${LOG_PATH}/${FILE_NAME}-text.log</file>
+        <encoder>
+            <!--            |%X{ru.nbch.scoring.web.logging.mdc_key}-->
+            <Pattern>${FILE_LOG_PATTERN}</Pattern>
+            <charset>utf8</charset>
+        </encoder>
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <fileNamePattern>${LOG_PATH}/${FILE_NAME}-text.%d{yyyy-MM-dd}.%i.gz
+            </fileNamePattern>
+            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+                <maxFileSize>100MB</maxFileSize>
+            </timeBasedFileNamingAndTriggeringPolicy>
+            <maxHistory>10</maxHistory>
+        </rollingPolicy>
+    </appender>
+    <!-- second FILE JSON appender -->
+    <appender name="JSON_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>${LOG_PATH}/${FILE_NAME}-json.log</file>
+        <encoder class="net.logstash.logback.encoder.LogstashEncoder" />
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <fileNamePattern>${LOG_PATH}/${FILE_NAME}-json.%d{yyyy-MM-dd}.%i.gz
+            </fileNamePattern>
+            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+                <maxFileSize>100MB</maxFileSize>
+            </timeBasedFileNamingAndTriggeringPolicy>
+            <maxHistory>10</maxHistory>
+        </rollingPolicy>
+    </appender>
+
+    <root level="info">
+         <appender-ref ref="CONSOLE"/>
+        <appender-ref ref="TEXT_FILE"/>
+        <appender-ref ref="JSON_FILE" />
+    </root>
+
+    <logger name="ru.spcex" level="debug" additivity="false">
+        <appender-ref ref="TEXT_FILE"/>
+        <appender-ref ref="JSON_FILE" />
+        <!-- <appender-ref ref="CONSOLE"/> -->
+    </logger>
+</configuration>

clearing-parent/control-service/src/test/java/ru/spcex/clearing/control/config/TestDataSourceConfiguration.java

@@ -0,0 +1,52 @@
+package ru.spcex.clearing.control.config;
+
+import com.mchange.v2.c3p0.ComboPooledDataSource;
+import java.beans.PropertyVetoException;
+import javax.sql.DataSource;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.testcontainers.containers.PostgreSQLContainer;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+
+@Configuration
+public class TestDataSourceConfiguration {
+    private final ControlServiceSettings settings;
+
+    public TestDataSourceConfiguration(ControlServiceSettings settings) {
+        this.settings = settings;
+    }
+
+    @Bean
+    public DataSource getDataSource() {
+        PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(PostgreSQLContainer.IMAGE);
+        postgres.start();
+
+        ComboPooledDataSource cpds = new ComboPooledDataSource();
+        try {
+            cpds.setDriverClass(postgres.getDriverClassName());
+            cpds.setJdbcUrl(postgres.getJdbcUrl());
+            cpds.setUser(postgres.getUsername());
+            cpds.setPassword(postgres.getPassword());
+
+            cpds.setMaxPoolSize(100);
+            cpds.setMinPoolSize(50);
+            cpds.setAcquireIncrement(5);
+        } catch (PropertyVetoException e) {
+            throw new RuntimeException(e);
+        }
+
+        return cpds;
+    }
+
+    @Bean
+    public JdbcTemplate getJdbcTemplate() {
+        return new JdbcTemplate(getDataSource());
+    }
+
+    @Bean
+    public NamedParameterJdbcTemplate getNamedParameterJdbcTemplate() {
+        return new NamedParameterJdbcTemplate(getDataSource());
+    }
+}

clearing-parent/control-service/src/test/java/ru/spcex/clearing/control/service/ControlServiceTest.java

@@ -0,0 +1,171 @@
+package ru.spcex.clearing.control.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.opencsv.CSVReader;
+import com.opencsv.exceptions.CsvException;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.Reader;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.jdbc.core.JdbcTemplate;
+import ru.spcex.clearing.control.config.TestDataSourceConfiguration;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+import ru.spcex.clearing.control.mapper.ResultContainerMapper;
+import ru.spcex.clearing.control.model.ResultContainer;
+import ru.spcex.clearing.control.repository.ControlRepository;
+import ru.spcex.platform.enumeration.SessionType;
+
+@SpringBootTest(classes = {
+    TestDataSourceConfiguration.class,
+    ControlService.class,
+    ControlRepository.class,
+    ResultContainerMapper.class,
+    ControlServiceSettings.class
+}, properties = {
+    "control-service.result-directory=src/test/resources"
+})
+@EnableConfigurationProperties
+class ControlServiceTest {
+    @Autowired
+    private ControlService controlService;
+    @Autowired
+    private JdbcTemplate jdbcTemplate;
+
+    @BeforeEach
+    void setUp() {
+        jdbcTemplate.execute("DROP TABLE IF EXISTS s_trades");
+        jdbcTemplate.execute("DROP TABLE IF EXISTS execution_deposit");
+        jdbcTemplate.execute("DROP TABLE IF EXISTS execution_fond");
+        jdbcTemplate.execute("DROP TABLE IF EXISTS execution_currency");
+        jdbcTemplate.execute("DROP TABLE IF EXISTS payment_instruction");
+        jdbcTemplate.execute("DROP TABLE IF EXISTS s_df03");
+
+        jdbcTemplate.execute("create table s_trades(id bigint not null primary key, created_at timestamp, updated_at timestamp, trade_num bigint, operation varchar(40), class_code varchar(255), trade_date date, sec_code varchar(255), accruedint numeric(72, 18), accruedint2 numeric(72, 18), lower_discount numeric(72, 18), order_num bigint, price numeric(72, 18), price2 numeric(72, 18), repo_rate numeric(72, 18), repo_value numeric(72, 18), repo2_value numeric(72, 18), start_discount numeric(72, 18), ts_commission numeric(72, 18), upper_discount numeric(72, 18), value numeric(72, 18), yield numeric(72, 18), qty numeric(72, 18), qty_pcs numeric(72, 18), trade_date_time timestamp, repo_term bigint, clearing_commission numeric(72, 18), exchange_commission numeric(72, 18), tech_center_commission numeric(72, 18), account varchar(50), broker_ref varchar(34), client_code varchar(255), settle_code varchar(50), user_id varchar(32), exchange_code varchar(64), firm_id varchar(255), firm_name varchar(255), cp_firm_id varchar(255), cp_firm_name varchar(255), class_name varchar(255), sec_name varchar(255), settle_date date, settle_currency varchar(4), trade_currency varchar(4), trade_time_ms bigint, bank_acc_id varchar(12), section varchar(4));");
+        jdbcTemplate.execute("create table execution_deposit(exchange_execution_id bigint,exchange_execution_time timestamp,trading_date date,trading_clearing_registry_id bigint,party_trading_clearing_registry varchar(20),market varchar(8),price numeric(72, 18),lots numeric(72, 2),quantity numeric(72, 2),first_leg_amount numeric(72, 2),second_leg_amount numeric(72, 2),interest_amount numeric(72, 2),side varchar(4),settlement_currency varchar(4),company_id bigint,duration bigint,first_leg_settlement_date date,second_leg_settlement_date date,first_leg_settlement_code varchar(12),second_leg_settlement_code varchar(12),security_full_name varchar(255),security_symbol varchar(255),security_code varchar(255),security_id bigint,contract varchar(255),counter_party_id bigint,counter_party_trading_clearing_registry_id bigint,counter_party_trading_clearing_registry varchar(20),coverage_status varchar(4),session_id bigint,id bigint not null primary key, created_at timestamp, updated_at timestamp, clearing_date date);");
+        jdbcTemplate.execute("create table execution_fond(id bigint not null primary key, created_at timestamp, updated_at timestamp, clearing_date date, exchange_execution_id bigint, side varchar(4), market varchar(4), trading_date date, security_symbol varchar(255), security_id bigint, interest_amount numeric(72, 2), exchange_order_id bigint, price numeric(72, 18), settlement_amount numeric(72, 2), lots numeric(72, 2), quantity numeric(72, 2), exchange_execution_time timestamp, duration bigint, trading_clearing_registry_id bigint, party_trading_clearing_registry varchar(20), comment varchar(255), client_code_id bigint, settlement_code varchar(12), company_id bigint, counter_party_id bigint, counter_party_trading_clearing_registry_id bigint, counter_party_trading_clearing_registry varchar(20), security_full_name varchar(255), settlement_date date, settlement_currency varchar(4), exchange_execution_microseconds timestamp, coverage_status varchar(4), session_id bigint);");
+        jdbcTemplate.execute("create table execution_currency( exchange_execution_id bigint, exchange_execution_time timestamp, exchange_execution_microseconds timestamp, trading_date date, settlement_date date, settlement_code varchar(12), security_id bigint, security_symbol varchar(255), security_name varchar(255), company_id bigint, party_trading_clearing_registry_id bigint, party_trading_clearing_registry varchar(20), counter_party_id bigint, counter_party_trading_clearing_registry_id bigint, counter_party_trading_clearing_registry varchar(20), market varchar(4), price numeric(72, 18), lots numeric(72, 2), settlement_amount numeric(72, 2), quantity numeric(72, 2), side varchar(4), currency_code varchar(4), settlement_organization varchar(255), coverage_status varchar(4), session_id bigint, clearing_date date, id bigint not null primary key, created_at timestamp, updated_at timestamp);");
+        jdbcTemplate.execute("create table payment_instruction( sender_id bigint, addressee_id bigint, payee_bic varchar(255), adressee_bic varchar(255), payee_bank_name varchar(255), addressee_bank_name varchar(255), payment_date timestamp, payment_purpose varchar(255), settlement_date date, credit_leg_amount numeric(72, 18), debit_leg_amount numeric(72, 18), credit_leg_account_id bigint, credit_cs_account varchar(255), credit_leg_account varchar(50), debit_leg_account_id bigint, debit_cs_account varchar(255), debit_leg_account varchar(50), credit_leg_direction varchar(4), debit_leg_direction varchar(4), credit_leg_currency_code varchar(4), debit_leg_currency_code varchar(4), credit_leg_security_id bigint, debit_leg_security_id bigint, transaction_status varchar(4), document_number varchar(255), id bigint not null primary key, created_at timestamp, updated_at timestamp, clearing_date date, session_id bigint);");
+        jdbcTemplate.execute("create table s_df03( id bigint not null primary key, seg_type varchar(1), doc_type varchar(4), docnm_ref varchar(16), docnmprev varchar(16), c_acc_deb varchar(35), sbanknam1 varchar(35), sbanknam2 varchar(35), sbanknam3 varchar(35), sbanknam4 varchar(35), sbanknam5 varchar(35), c_acc_cred varchar(35), rbanknam1 varchar(35), rbanknam2 varchar(35), rbanknam3 varchar(35), rbanknam4 varchar(35), rbanknam5 varchar(35), pay_date varchar(8), pay_val varchar(12), sum_deb varchar(22), specif_1 varchar(255), imp_result varchar(3), file_name varchar(255), generation_time timestamp, generation_id bigint, payment_instruction_id bigint);");
+
+        jdbcTemplate.execute("insert into s_trades (id, created_at, section) values (1, NOW(), 'FINL'), (2, NOW(), 'FOND'), (3, NOW(), 'CURR');");
+        jdbcTemplate.execute("insert into execution_deposit (id, created_at) values (1, NOW()), (2, NOW());");
+        jdbcTemplate.execute("insert into execution_fond (id, created_at) values (1, NOW()), (2, NOW());");
+        jdbcTemplate.execute("insert into execution_currency (id, created_at) values (1, NOW()), (2, NOW());");
+        jdbcTemplate.execute("insert into payment_instruction (id, created_at, session_id) values (1, NOW(), 1), (2, NOW(), 1);");
+        jdbcTemplate.execute("insert into s_df03 (id, generation_time, payment_instruction_id) values (1, NOW(), 1), (2, NOW(), 1);");
+    }
+
+    @Test
+    void generalControl_shouldReturnFINLCSVFile() {
+        ResultContainer resultContainer = new ResultContainer(null, SessionType.FINL, "s_trades", 3L, "execution_deposit", 2L, 1L);
+
+        File resultFile = controlService.generalControl(SessionType.FINL);
+        assertThat(resultFile).isNotNull();
+
+        ResultContainer actual;
+        try (Reader reader = new FileReader(resultFile);
+             CSVReader csvReader = new CSVReader(reader)) {
+            actual = parseResultContainer(csvReader.readAll());
+        } catch (IOException | CsvException e) {
+            throw new RuntimeException(e);
+        }
+        resultFile.delete();
+
+        assertThat(actual).isEqualTo(resultContainer);
+    }
+
+    @Test
+    void generalControl_shouldReturnIPOBCSVFile() {
+        ResultContainer resultContainer = new ResultContainer(null, SessionType.IPOB, "s_trades", 1L, "execution_fond", 2L, -1L);
+
+        File resultFile = controlService.generalControl(SessionType.IPOB);
+        assertThat(resultFile).isNotNull();
+
+        ResultContainer actual;
+        try (Reader reader = new FileReader(resultFile);
+             CSVReader csvReader = new CSVReader(reader)) {
+            actual = parseResultContainer(csvReader.readAll());
+        } catch (IOException | CsvException e) {
+            throw new RuntimeException(e);
+        }
+        resultFile.delete();
+
+        assertThat(actual).isEqualTo(resultContainer);
+    }
+
+    @Test
+    void generalControl_shouldReturnCURRCSVFile() {
+        ResultContainer resultContainer = new ResultContainer(null, SessionType.CURR, "s_trades", 1L, "execution_currency", 2L, -1L);
+
+        File resultFile = controlService.generalControl(SessionType.CURR);
+        assertThat(resultFile).isNotNull();
+
+        ResultContainer actual;
+        try (Reader reader = new FileReader(resultFile);
+             CSVReader csvReader = new CSVReader(reader)) {
+            actual = parseResultContainer(csvReader.readAll());
+        } catch (IOException | CsvException e) {
+            throw new RuntimeException(e);
+        }
+        resultFile.delete();
+
+        assertThat(actual).isEqualTo(resultContainer);
+    }
+
+    @Test
+    void internalControl_shouldReturnInternalControlCSVFile() {
+        ResultContainer resultContainer = new ResultContainer(1L, null, null, 1L, null, 2L, 3L);
+
+        File resultFile = controlService.internalControl(1L);
+        assertThat(resultFile).isNotNull();
+
+        ResultContainer actual;
+        try (Reader reader = new FileReader(resultFile);
+             CSVReader csvReader = new CSVReader(reader)) {
+            actual = parseResultContainer(csvReader.readAll());
+        } catch (IOException | CsvException e) {
+            throw new RuntimeException(e);
+        }
+        resultFile.delete();
+
+        assertThat(actual).isEqualTo(resultContainer);
+    }
+
+    @Test
+    void internalControlFinl_shouldReturnInternalControlCSVFile() {
+        ResultContainer resultContainer = new ResultContainer(1L, null, "payment_instruction", 2L, "s_df03", 3L, -1L);
+
+        File resultFile = controlService.internalControlFinl(1L);
+        assertThat(resultFile).isNotNull();
+
+        ResultContainer actual;
+        try (Reader reader = new FileReader(resultFile);
+             CSVReader csvReader = new CSVReader(reader)) {
+            actual = parseResultContainer(csvReader.readAll());
+        } catch (IOException | CsvException e) {
+            throw new RuntimeException(e);
+        }
+        resultFile.delete();
+
+        assertThat(actual).isEqualTo(resultContainer);
+    }
+
+    private ResultContainer parseResultContainer(List<String[]> rows) {
+        ResultContainer actual = new ResultContainer();
+        actual.setSessionId(rows.get(1)[0].isBlank() ? null : Long.valueOf(rows.get(1)[0]));
+        actual.setSessionType(rows.get(1)[1].isBlank() ? null : SessionType.valueOf(rows.get(1)[1]));
+        actual.setFirstTableTitle(rows.get(1)[2].isBlank() ? null : rows.get(1)[2]);
+        actual.setFirstTableRecordQuantity(rows.get(1)[3].isBlank() ? null : Long.valueOf(rows.get(1)[3]));
+        actual.setSecondTableRecordTitle(rows.get(1)[4].isBlank() ? null : rows.get(1)[4]);
+        actual.setSecondTableRecordQuantity(rows.get(1)[5].isBlank() ? null : Long.valueOf(rows.get(1)[5]));
+        actual.setQuantityDifference(rows.get(1)[6].isBlank() ? null : Long.valueOf(rows.get(1)[6]));
+        return actual;
+    }
+}

clearing-parent/pom.xml

@@ -45,6 +45,7 @@
         <module>imdg-hist</module>
         <module>xml-importer</module>
         <module>xml-exporter</module>
+        <module>control-service</module>
     </modules>
 
     <properties>

platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java

@@ -59,6 +59,9 @@ public enum Task implements IEnumKey {
     finishBadSessions("CCLR"),//Завершение неудачных клиринговых сессий
     sendLim_LIMC("LIMC"), // Выгрузка в Торговую систему остатков по валюте (отправка lim)"
     makeFiles_MTCR("MTCR"), // Формирование файлов с МТКР
+    generalControl_GLCL("GLCL"), // Сверка получения сделок из ТС в КС
+    internalControl_ILCL("ILCL"), // Внутренняя сверка по итогу формирования распоряжений на фондовой секции
+    internalFinlControl_IFCL("IFCL"), // Внутренняя сверка по итогу формирования распоряжений на секции МКР
     ;
 
     private final String key;

pom.xml

@@ -55,6 +55,7 @@
         <folder_root_gateway-api>${folder_root_clearing}/clearing-parent/gateway-api</folder_root_gateway-api>
         <folder_root_xml-exporter>${folder_root_clearing}/clearing-parent/xml-exporter</folder_root_xml-exporter>
         <folder_root_xml-importer>${folder_root_clearing}/clearing-parent/xml-importer</folder_root_xml-importer>
+        <folder_root_control-service>${folder_root_clearing}/clearing-parent/control-service</folder_root_control-service>
         <!-- IMDG -->
         <external_libraries.hazelcast.version>3.12.4</external_libraries.hazelcast.version>
         <external_libraries.slf4j.version>1.7.33</external_libraries.slf4j.version>

z-distr/pom.xml

@@ -614,6 +614,29 @@
                                     </fileSets>
                                 </configuration>
                             </execution>
+                            <execution>
+                                <id>copy-control-service-bin</id>
+                                <phase>prepare-package</phase>
+                                <goals>
+                                    <goal>copy</goal>
+                                </goals>
+                                <configuration>
+                                    <fileSets>
+                                        <fileSet>
+                                            <sourceFile>${folder_root_control-service}/target/control-service.jar</sourceFile>
+                                            <destinationFile>${folder.clearing.distr.bin}/control-service.jar</destinationFile>
+                                        </fileSet>
+                                        <fileSet>
+                                            <sourceFile>${folder_root_control-service}/target/control-service.jar</sourceFile>
+                                            <destinationFile>${folder.clearing.distr.services}/control-service/control-service.jar</destinationFile>
+                                        </fileSet>
+                                        <fileSet>
+                                            <sourceFile>${folder_root_control-service}/src/main/resources/application.properties</sourceFile>
+                                            <destinationFile>${folder.clearing.distr.settings}/control-service/application.properties</destinationFile>
+                                        </fileSet>
+                                    </fileSets>
+                                </configuration>
+                            </execution>
                         </executions>
                     </plugin>
                     <plugin>

z-distr/src/main/resources/distr/sh/control-service.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+CLEARING_HOME=/opt/mfd/clearing/
+cd $CLEARING_HOME/bin
+
+CMD="java -jar control-service.jar --spring.config.location=$CLEARING_HOME/settings/control-service/"
+
+$CMD >/dev/null 2>&1 &
+