Merge remote-tracking branch 'origin/dev' into dev

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/validation/ClientCodeValidationConfig.java

@@ -1,4 +1,4 @@
-package ru.spcex.clearing.company.config.validation;
+package ru.spcex.clearing.account.config.validation;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -7,7 +7,7 @@ import ru.clearing.classes.statics.data.account.ClientCode;
 import ru.clearing.classes.statics.data.company.Company;
 import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
 import ru.clearing.platform.dictionary.WorkflowStatusDictionary;
-import ru.spcex.clearing.company.error.CompanyErrors;
+import ru.spcex.clearing.account.errors.AccountError;
 import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeUpdateRequest;
@@ -40,33 +40,33 @@ public class ClientCodeValidationConfig {
             addImdg.accept(IMDGDistributedNames.Map_TradingClearingRegistry);
             addImdg.accept(IMDGDistributedNames.Map_WorkflowStatusDictionary);
             return new ValidatorImpl<>(context,
-                    FieldRequiredRule.instance("companyId", ClientCodeNewRequest::getCompanyId, CompanyErrors.RequiredFieldEmpty),
+                    FieldRequiredRule.instance("companyId", ClientCodeNewRequest::getCompanyId, AccountError.RequiredFieldEmpty),
                     IdPresentRule.instance("companyId",
                             ClientCodeNewRequest::getCompanyId,
                             IMDGDistributedNames.Map_Company,
                             Company.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.CompanyNotFound),
+                            AccountError.CompanyNotFound),
 
                     IdPresentRule.instance("moneyAccountId",
                             ClientCodeNewRequest::getMoneyAccountId,
                             IMDGDistributedNames.Map_Account,
                             Account.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.AccountNotFound,
+                            AccountError.AccountNotFound,
                             false,
                             acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
-                                    ? null : CompanyErrors.AccountNotFound
+                                    ? null : AccountError.AccountNotFound
                             ),
                     IdPresentRule.instance("depoAccountId",
                             ClientCodeNewRequest::getDepoAccountId,
                             IMDGDistributedNames.Map_Account,
                             Account.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.AccountNotFound,
+                            AccountError.AccountNotFound,
                             false,
                             acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
-                                    ? null : CompanyErrors.AccountNotFound
+                                    ? null : AccountError.AccountNotFound
                     ),
 
 
@@ -74,8 +74,8 @@ public class ClientCodeValidationConfig {
                             ClientCodeNewRequest::getStatus,
                             IMDGDistributedNames.Map_WorkflowStatusDictionary,
                             WorkflowStatusDictionary.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.DictionaryNotFound,
+                            AccountError.DictionaryNotFound,
                             false)
             );
         };
@@ -97,45 +97,45 @@ public class ClientCodeValidationConfig {
                             ClientCodeUpdateRequest::getId,
                             IMDGDistributedNames.Map_ClientCode,
                             ClientCode.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.RecordNotFound
+                            AccountError.RecordNotFound
                     ),
 
-                    FieldRequiredRule.instance("companyId", ClientCodeUpdateRequest::getCompanyId, CompanyErrors.RequiredFieldEmpty),
+                    FieldRequiredRule.instance("companyId", ClientCodeUpdateRequest::getCompanyId, AccountError.RequiredFieldEmpty),
                     IdPresentRule.instance("companyId",
                             ClientCodeUpdateRequest::getCompanyId,
                             IMDGDistributedNames.Map_Company,
                             Company.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.CompanyNotFound),
+                            AccountError.CompanyNotFound),
 
                     IdPresentRule.instance("moneyAccountId",
                             ClientCodeUpdateRequest::getMoneyAccountId,
                             IMDGDistributedNames.Map_Account,
                             Account.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.AccountNotFound,
+                            AccountError.AccountNotFound,
                             false,
                             acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
-                                    ? null : CompanyErrors.AccountNotFound
+                                    ? null : AccountError.AccountNotFound
                     ),
                     IdPresentRule.instance("depoAccountId",
                             ClientCodeUpdateRequest::getDepoAccountId,
                             IMDGDistributedNames.Map_Account,
                             Account.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.AccountNotFound,
+                            AccountError.AccountNotFound,
                             false,
                             acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
-                                    ? null : CompanyErrors.AccountNotFound
+                                    ? null : AccountError.AccountNotFound
                     ),
 
                     DictionaryPresentRule.instance("stauts",
                             ClientCodeUpdateRequest::getStatus,
                             IMDGDistributedNames.Map_WorkflowStatusDictionary,
                             WorkflowStatusDictionary.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.DictionaryNotFound,
+                            AccountError.DictionaryNotFound,
                             false)
             );
         };
@@ -154,8 +154,8 @@ public class ClientCodeValidationConfig {
                             CommonDeleteRequest::getId,
                             IMDGDistributedNames.Map_ClientCode,
                             ClientCode.class,
-                            CompanyErrors.RequiredFieldEmpty,
+                            AccountError.RequiredFieldEmpty,
-                            CompanyErrors.RecordNotFound)
+                            AccountError.RecordNotFound)
             );
         };
     }

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/validation/ValidationConfig.java

@@ -7,10 +7,7 @@ import ru.clearing.classes.statics.data.account.*;
 import ru.clearing.classes.statics.data.company.ClearingMemberCategory;
 import ru.clearing.classes.statics.data.company.Company;
 import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
-import ru.clearing.platform.dictionary.AccountTypeDictionary;
+import ru.clearing.platform.dictionary.*;
-import ru.clearing.platform.dictionary.ClearingAccountTypeDictionary;
-import ru.clearing.platform.dictionary.CurrencyCodeDictionary;
-import ru.clearing.platform.dictionary.ServiceStatusDictionary;
 import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.validation.common.ValidationHelper;
 import ru.spcex.platform.classes.base.SpcexObjectBase;
@@ -43,6 +40,10 @@ public class ValidationConfig {
         addImdg.accept(IMDGDistributedNames.Map_ServiceStatusDictionary, ServiceStatusDictionary.class);
         addImdg.accept(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class);
 
+        //for ClientCodeValidationConfig
+        addImdg.accept(IMDGDistributedNames.Map_WorkflowStatusDictionary, WorkflowStatusDictionary.class);
+        addImdg.accept(IMDGDistributedNames.Map_ClientCode, ClientCode.class);
+
         return imdg;
     }
 

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/errors/AccountError.java

@@ -6,7 +6,9 @@ public enum AccountError implements IErrorEnumId {
     GeneralError(5000L),
     UserVerifyDenial(5001L),
     RequiredFieldEmpty(5002L),
+    DictionaryNotFound(5003L),
     WrongFieldValue(5004L),
+    RecordNotFound(5006L),
     AccountAlreadyExist(5010L),
     AccountNotFound(5011L),
     AccountNotActive(5012L),

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/ClientCodeService.java

@@ -1,4 +1,4 @@
-package ru.spcex.clearing.company.service;
+package ru.spcex.clearing.account.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.producer.Producer;
@@ -14,7 +14,7 @@ import org.springframework.stereotype.Service;
 import ru.clearing.classes.statics.data.account.ClientCode;
 import ru.clearing.classes.statics.data.company.Company;
 import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
-import ru.spcex.clearing.company.error.CompanyErrors;
+import ru.spcex.clearing.account.errors.AccountError;
 import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.domain.ActionType;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
@@ -26,8 +26,10 @@ import ru.spcex.clearing.platform.messaging.domain.cud.registry.TradingClearingR
 import ru.spcex.clearing.platform.messaging.domain.cud.registry.TradingClearingRegistryUpdateRequest;
 import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
 import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
+import ru.spcex.clearing.platform.messaging.service.Status;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
 import ru.spcex.clearing.util.security.UserRoleVerification;
+import ru.spcex.clearing.util.services.RequestHelper;
 import ru.spcex.clearing.util.services.exchangers.BiDirectionQueueExchanger;
 import ru.spcex.clearing.validation.common.ValidationHelper;
 import ru.spcex.platform.enumeration.TradingClearingRegistryType;
@@ -51,12 +53,11 @@ import java.util.concurrent.Future;
 import java.util.function.Function;
 
 @Service
-public class ClientCodeService extends QueueConsumer implements InitializingBean, DisposableBean {
+public class ClientCodeService extends QueueConsumer implements InitializingBean {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Producer<String, Object> kafkaProducer;
     private final ImdgId idGenerator;
     private final Imdg<ClientCode> clientCodeMap;
-    private final Imdg<Company> companyMap;
     private final Imdg<TradingClearingRegistry> tradingClearingRegistryMap;
 
 
@@ -66,37 +67,35 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
     private final ValidationHelper validationHelper;
     private final UserRoleVerification userRoleVerification;
     private final IMessageResolver messageResolver;
+    private final RequestHelper requestHelper;
 
-    BiDirectionQueueExchanger<BaseRequest<TradingClearingRegistryNewRequest>> accountServiceExchanger;
+    protected TradingClearingRegistryService tradingClearingRegistryService;
 
     @Autowired
-    public ClientCodeService(Consumer<String, Object> kafkaQueue1, Consumer<String, Object> kafkaQueue2,
+    public ClientCodeService(Consumer<String, Object> kafkaQueue,
                              Producer<String, Object> kafkaProducer,
                              ImdgProvider imdgProvider,
                              ValidationHelper validationHelper,
                              IMessageResolver messageResolver,
+                             RequestHelper requestHelper,
+                             TradingClearingRegistryService tradingClearingRegistryService,
                              UserRoleVerification userRoleVerification,
                              @Qualifier("clientCodeNewRequestValidator") Function<ClientCodeNewRequest, IValidator> clientCodeNewRequestValidator,
                              @Qualifier("clientCodeUpdateRequestValidator") Function<ClientCodeUpdateRequest, IValidator> clientCodeUpdateRequestValidator,
                              @Qualifier("clientCodeDeleteRequestValidator") Function<CommonDeleteRequest, IValidator> clientCodeDeleteRequestValidator) {
-        super(kafkaQueue1, kafkaProducer);
+        super(kafkaQueue, kafkaProducer);
         this.kafkaProducer = kafkaProducer;
         this.idGenerator = imdgProvider.getImdgIdGenerator();
         this.clientCodeMap = imdgProvider.getImdg(IMDGDistributedNames.Map_ClientCode, ClientCode.class);
-        this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class);
         this.tradingClearingRegistryMap = imdgProvider.getImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class);
         this.clientCodeNewRequestValidator = clientCodeNewRequestValidator;
         this.clientCodeUpdateRequestValidator = clientCodeUpdateRequestValidator;
         this.clientCodeDeleteRequestValidator = clientCodeDeleteRequestValidator;
         this.validationHelper = validationHelper;
+        this.requestHelper = requestHelper;
         this.messageResolver = messageResolver;
         this.userRoleVerification = userRoleVerification;
-
+        this.tradingClearingRegistryService = tradingClearingRegistryService;
-        this.accountServiceExchanger = new BiDirectionQueueExchanger<>(kafkaQueue2, kafkaProducer,
-                Consts.DESTINATION_TRADING_CLEARING_REGISTRY_NEW,
-                Consts.DESTINATION_TRADING_CLEARING_REGISTRY_REPLY, true, // REQUEST_INFO_UPDATE - стандартная очередь, для результатов всех реквестов. DESTINATION_TRADING_CLEARING_REGISTRY_REPLY,
-                60000
-        );
     }
 
     @Override
@@ -118,10 +117,6 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
 
     }
 
-    @Override
-    public void destroy() {
-        accountServiceExchanger.close();
-    }
 
     protected RequestInfoUpdate clientCodeNew(BaseRequest<ClientCodeNewRequest> userRequest) {
         log.debug("ClientCodeNewRequest received {}", userRequest.getId());
@@ -143,7 +138,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
                 log.error("Can not wait creation of TCR. request id={};CompanyId={}, MoneyAccountId={}, DepoAccountId={}; {}",
                         userRequest.getId(), req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId(),
                         e.toString());
-                return makeErrorResponse(userRequest, CompanyErrors.GeneralError, "Can not create TCR at this moment");
+                return makeErrorResponse(userRequest, AccountError.GeneralError, "Can not create TCR: " + e.getMessage());
             }
         }
 
@@ -176,7 +171,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
                 log.error("Can not wait creation of TCR. request id={};CompanyId={}, MoneyAccountId={}, DepoAccountId={}; {}",
                         userRequest.getId(), req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId(),
                         e.toString());
-                return makeErrorResponse(userRequest, CompanyErrors.GeneralError, "Can not create TCR at this moment");
+                return makeErrorResponse(userRequest, AccountError.GeneralError, "Can not create TCR at this moment" + e.getMessage());
             }
         }
 
@@ -201,7 +196,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
         log.debug("ClientCodeUpdateRequest received, id={}", req.getId());
         ClientCode clientCode = clientCodeMap.getSingleObjectByID(req.getId());
         if (clientCode == null) {
-            return makeErrorResponse(userRequest, CompanyErrors.RecordNotFound, req.getId());
+            return makeErrorResponse(userRequest, AccountError.RecordNotFound, req.getId());
         }
 
         boolean doCreateTCR = checkNeedCreateTCR(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
@@ -213,7 +208,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
                 log.error("Can not wait creation of TCR. request id={};CompanyId={}, MoneyAccountId={}, DepoAccountId={}; {}",
                         userRequest.getId(), req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId(),
                         e.toString());
-                return makeErrorResponse(userRequest, CompanyErrors.GeneralError, "Can not create TCR at this moment");
+                return makeErrorResponse(userRequest, AccountError.GeneralError, "Can not create TCR: " + e.getMessage());
             }
         }
 
@@ -224,13 +219,8 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
         return null;
     }
 
-    private RequestInfoUpdate makeErrorResponse(BaseRequest<?> req, CompanyErrors err, Object... arg) {
+    private RequestInfoUpdate makeErrorResponse(BaseRequest<?> req, AccountError err, Object... arg) {
-        String errMsg = messageResolver.resolve(new EnumMessage(err, arg));
+        return requestHelper.makeErrorResponse(req, err, arg);
-        return new RequestInfoUpdate()
-                .setId(req.getId())
-                .setStatus(ru.spcex.clearing.platform.messaging.service.Status.Error)
-                .setMessage(errMsg);
-
     }
 
     protected RequestInfoUpdate clientCodeDelete(BaseRequest<CommonDeleteRequest> userRequest) {
@@ -296,14 +286,15 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
         requestPayload.setTradingClearingRegistryType(TradingClearingRegistryType.Client_B.getKey());
         request.setRequestPayload(requestPayload);
         try {
-            Long reply = accountServiceExchanger.exchange(request);
+            // Следующий вызываемый метод обязательно должен быть synchronized.
-            if (reply == null) {
+            RequestInfoUpdate reply = tradingClearingRegistryService.tradingClearingRegistryNew(request);
-                throw new ClearingBaseException(CompanyErrors.GeneralError, "Waiting account-service timeout");
+            if (reply != null && Status.Error.equals(reply.getStatus())) {
+                throw new ClearingBaseException(AccountError.GeneralError, "tradingClearingRegistryService return error: " + reply.getMessage());
             }
-            // примечание: ошибка и сбой (непредвиденное завершение программы) не приведёт к необратимым последствиям,
+        } catch (ClearingBaseException expected) {
-            // т.к. пользователь сможет повторить запрос, а созданный на предыдущем запросе ТКР уже будет создан и найдётся.
+            throw expected;
-        } catch (InterruptedException e) {
+        } catch (Exception e) {
-            throw new ClearingBaseException(CompanyErrors.GeneralError, "Waiting account-service timeout");
+            throw new ClearingBaseException(AccountError.GeneralError, "Waiting account-service timeout");
         }
     }
 

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/TradingClearingRegistryService.java

@@ -272,7 +272,7 @@ public class TradingClearingRegistryService extends QueueConsumer implements Ini
         return null;
     }
 
-    public RequestInfoUpdate tradingClearingRegistryNew(BaseRequest<TradingClearingRegistryNewRequest> userRequest) {
+    public synchronized RequestInfoUpdate tradingClearingRegistryNew(BaseRequest<TradingClearingRegistryNewRequest> userRequest) {
         log.debug("TradingClearingRegistryNewRequest received");
         RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(userRequest);
         if (requestInfoUpdate != null) {
@@ -328,39 +328,9 @@ public class TradingClearingRegistryService extends QueueConsumer implements Ini
         tradingClearingRegistryImdg.insert(tradingClearingRegistry);
 
         log.debug("successfully processed, id {}", id);
-
-        sendResponse(userRequest, null, null); // to DESTINATION_TRADING_CLEARING_REGISTRY_REPLY
-
         return null;
     }
 
-
-    private void sendResponse(BaseRequest<?> o, Object response, Header correlationId) {
-            try {
-                Future<RecordMetadata> send;
-                BaseRequest<Object> req = new BaseRequest<>();
-                req.setId(o.getId());
-                req.setActionType(ActionType.SYSTEM);
-                if (response != null) {
-                    req.setRequestPayload(response);
-                } else {
-                    //default response
-                    RequestInfoUpdate success = new RequestInfoUpdate();
-                    success.setId(o.getId());
-                    success.setStatus(Status.Success);
-                    req.setRequestPayload(success);
-                }
-                ProducerRecord<String, Object> respRec = new ProducerRecord<>(Consts.DESTINATION_TRADING_CLEARING_REGISTRY_REPLY, req);
-                if (correlationId != null) {
-                    respRec.headers().add(new CorrelationHeader(correlationId.value().clone()));
-                }
-                send = kafkaProducer.send(respRec);
-                send.get();
-            } catch (Exception e) {
-                log.error(ExceptionUtils.getStackTrace(e));
-            }
-    }
-
     public RequestInfoUpdate tradingClearingRegistryUpdate(BaseRequest<TradingClearingRegistryUpdateRequest> userRequest) {
         log.debug("TradingClearingRegistryUpdateRequest received");
         RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(userRequest);

clearing-parent/account-service/src/test/java/ru/spcex/clearing/account/service/ClientCodeServiceTest.java

@@ -1,4 +1,4 @@
-package ru.spcex.clearing.company.service;
+package ru.spcex.clearing.account.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.producer.MockProducer;
@@ -8,58 +8,62 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
+import org.mockito.Mockito;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.account.Account;
+import ru.clearing.classes.statics.data.account.ClearingAccount;
 import ru.clearing.classes.statics.data.account.ClientCode;
+import ru.clearing.classes.statics.data.account.DepoAccount;
 import ru.clearing.classes.statics.data.company.Company;
 import ru.clearing.classes.statics.data.profile.CompanyInfo;
 import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
 import ru.clearing.platform.dictionary.*;
-import ru.spcex.clearing.company.config.BeanConfiguration;
+import ru.spcex.clearing.account.config.BeanConfiguration;
-import ru.spcex.clearing.company.config.validation.ClientCodeValidationConfig;
+import ru.spcex.clearing.account.config.HazelcastServiceTestConfiguration;
-import ru.spcex.clearing.company.config.validation.ValidationConfig;
+import ru.spcex.clearing.account.config.KafkaConfigTest;
+import ru.spcex.clearing.account.config.validation.ClientCodeValidationConfig;
+import ru.spcex.clearing.account.config.validation.TradingClearingRegistryValidationConfig;
+import ru.spcex.clearing.account.config.validation.ValidationConfig;
+import ru.spcex.clearing.account.utils.MatcherFactory;
+import ru.spcex.clearing.account.utils.TestUtils;
 import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeUpdateRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
-import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
-import ru.spcex.clearing.test.config.ImdgTestConfig;
-import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.TradingClearingRegistryType;
 import ru.spcex.platform.enumeration.WorkflowStatus;
 import ru.spcex.platform.imdg.api.Imdg;
-import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService;
 
 import javax.annotation.PostConstruct;
 
 import static org.junit.jupiter.api.Assertions.*;
-import static org.mockito.Mockito.*;
-import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
-import static ru.spcex.clearing.test.TestUtils.*;
-import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
 
 @ExtendWith(SpringExtension.class)
 @ContextConfiguration(classes = {
         ClientCodeService.class,
         ClientCodeValidationConfig.class,
 
+        TradingClearingRegistryService.class,
+        TradingClearingRegistryValidationConfig.class,
+
         ValidationConfig.class,
         BeanConfiguration.class,
 
-        KafkaTestConfig.class,
+        KafkaConfigTest.class,
-        ImdgTestConfig.class})
+
+        HazelcastServiceTestConfiguration.class,})
 class ClientCodeServiceTest {
 
     private static final int PARTITION = 0;
     private static final Long ID = 4L;
-    public static final MatcherFactory.Matcher<ClientCode> CLIENT_CODE_MATCHER = usingIgnoringFieldsComparator("created","updated");
+    public static final MatcherFactory.Matcher<ClientCode> CLIENT_CODE_MATCHER = MatcherFactory.usingIgnoringFieldsComparator("created", "updated");
 
     private static final Long TCR_ID = 41L;
     private static final Long COMPANY_ID = 42L;
@@ -68,7 +72,7 @@ class ClientCodeServiceTest {
     ClientCodeService clientCodeService;
     @Autowired
     @Qualifier("hazelcastServiceTest")
-    private ImdgProvider hazelcastServiceTest;
+    private HazelcastService hazelcastServiceTest;
     @Captor
     private ArgumentCaptor<ProducerRecord> producerRecord;
     @SpyBean
@@ -81,7 +85,7 @@ class ClientCodeServiceTest {
 
     @PostConstruct
     private void init() {
-        waitAvailableImdgProviderAndAddAdminWithDefaultId();
+        hazelcastServiceTest.waitAvailable();
         clientCodeImdg = hazelcastServiceTest.getImdg(IMDGDistributedNames.Map_ClientCode, ClientCode.class);
 
         // Словари для теста, применяются в ValidationConfig
@@ -120,21 +124,33 @@ class ClientCodeServiceTest {
         tcrImdg.insert(registry1);
 
         Imdg<Account> accounts = hazelcastServiceTest.getImdg(IMDGDistributedNames.Map_Account, Account.class);
-        Account moneyAccount=new Account();
+        Imdg<ClearingAccount> clsAccounts = hazelcastServiceTest.getImdg(IMDGDistributedNames.Map_ClearingAccount, ClearingAccount.class);
+        Account moneyAccount = new Account();
         moneyAccount.setId(131L);
         moneyAccount.setAccount("AAAA-4444");
         moneyAccount.setStatus("ACTV");
         moneyAccount.setCompanyId(COMPANY_ID); // для валидации принадлежности счёта
         accounts.insert(moneyAccount);
-        Account depoAccount=new Account();
+        ClearingAccount clsAcc = new ClearingAccount();
+        clsAcc.setId(moneyAccount.getId());
+        clsAcc.setCompanyId(COMPANY_ID);
+        clsAcc.setAccountId(moneyAccount.getId());
+        clsAccounts.insert(clsAcc);
+        Imdg<DepoAccount> depoAccounts = hazelcastServiceTest.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class);
+        Account depoAccount = new Account();
         depoAccount.setId(132L);
         depoAccount.setAccount("AAAB-44654");
         depoAccount.setStatus("ACTV");
         depoAccount.setCompanyId(COMPANY_ID); // для валидации принадлежности счёта
         accounts.insert(depoAccount);
+        DepoAccount depoAcc = new DepoAccount();
+        depoAcc.setId(depoAccount.getId());
+        depoAcc.setCompanyId(COMPANY_ID);
+        depoAcc.setAccountId(depoAccount.getId());
+        depoAccounts.insert(depoAcc);
 
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
+        TestUtils.FutureRecordMetadata future = Mockito.spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
+        Mockito.doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     private <D extends AbstractDictionary> void putToDictionary(String mapName, D object, String code) {
@@ -187,12 +203,12 @@ class ClientCodeServiceTest {
 //        predictableClientCode.setTradingClearingRegistryId(TCR_ID);
 
         //ACT
-        String jsonString = getJsonStringForNew(clientCodeNewRequest, ID);
+        String jsonString = TestUtils.getJsonStringForNew(clientCodeNewRequest, ID);
 
-        addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_NEW, PARTITION, 0, jsonString);
+        TestUtils.addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        TestUtils.waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
         ClientCode resultNew = clientCodeImdg.getSingleObjectBySQL(String.format("code = '%s'", ccCode));
         predictableClientCode.setId(resultNew.getId());
         CLIENT_CODE_MATCHER.assertMatch(resultNew, predictableClientCode);
@@ -225,19 +241,57 @@ class ClientCodeServiceTest {
 //        predictableClientCode.setTradingClearingRegistryId(TCR_ID);
 
         //ACT
-        String jsonString = getJsonStringForNew(clientCodeNewRequest, ID);
+        String jsonString = TestUtils.getJsonStringForNew(clientCodeNewRequest, ID);
 
-        addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_NEW_UM_COMPANY, PARTITION, 0, jsonString);
+        TestUtils.addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_NEW_UM_COMPANY, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        TestUtils.waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
         ClientCode resultNew = clientCodeImdg.getSingleObjectBySQL(String.format("code = '%s'", ccCode));
         predictableClientCode.setId(resultNew.getId());
         CLIENT_CODE_MATCHER.assertMatch(resultNew, predictableClientCode);
         assertNotNull(resultNew.getCreated());
     }
 
-    //todo добавить тест NEW  заполненными MoneyAccountId(131L), DepoAccountId(132L); - от этого направляется дополнительное сообщение в очередь и используется ожидание ответа.
+
+    /**
+     * {@link ClientCodeService#clientCodeUpdate(BaseRequest)}<br>
+     * Тест проверяет создание {@link ClientCode} в IMDG при передаче из Apache Kafka (очередь 1).<br>
+     * NEW с заполненными MoneyAccountId(131L), DepoAccountId(132L); - от этого будет создан ТКР.
+     * Входной запрос {@link ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeNewRequest}:<br>
+     **/
+    @Test
+    void clientCodeNew3() {
+        //ARRANGE
+        final String ccCode = "Lucky planet2";
+        ClientCodeNewRequest clientCodeNewRequest = new ClientCodeNewRequest();
+        clientCodeNewRequest.setCompanyId(COMPANY_ID);
+        clientCodeNewRequest.setCode(ccCode);
+        clientCodeNewRequest.setTradingClearingRegistryId(null);
+        clientCodeNewRequest.setMoneyAccountId(131L);
+        clientCodeNewRequest.setDepoAccountId(132L);
+        clientCodeNewRequest.setStatus("ACTV");
+
+        ClientCode predictableClientCode = new ClientCode();
+        predictableClientCode.setCode(ccCode);
+        predictableClientCode.setStatus("ACTV");
+        predictableClientCode.setCompanyId(COMPANY_ID);
+        predictableClientCode.setMoneyAccountId(131L);
+        predictableClientCode.setDepoAccountId(132L);
+        predictableClientCode.setTradingClearingRegistryId(TCR_ID);
+
+        //ACT
+        String jsonString = TestUtils.getJsonStringForNew(clientCodeNewRequest, ID);
+
+        TestUtils.addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_NEW, PARTITION, 0, jsonString);
+
+        //ASSERT
+        TestUtils.waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        ClientCode resultNew = clientCodeImdg.getSingleObjectBySQL(String.format("code = '%s'", ccCode));
+        predictableClientCode.setId(resultNew.getId());
+        CLIENT_CODE_MATCHER.assertMatch(resultNew, predictableClientCode);
+        assertNotNull(resultNew.getCreated());
+    }
 
     /**
      * {@link ClientCodeService#clientCodeUpdate(BaseRequest)}<br>
@@ -276,12 +330,12 @@ class ClientCodeServiceTest {
         predictableClientCode.setStatus("ACTV");
 
         //ACT
-        String jsonString = getJsonStringForUpdate(clientCodeUpdateRequest, ID);
+        String jsonString = TestUtils.getJsonStringForUPDATE(clientCodeUpdateRequest, ID);
 
-        addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_UPDATE, PARTITION, 0, jsonString);
+        TestUtils.addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        TestUtils.waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
 
         ClientCode resultUpdating = clientCodeImdg.getSingleObjectByID(ID);
         CLIENT_CODE_MATCHER.assertMatch(resultUpdating, predictableClientCode);
@@ -315,13 +369,13 @@ class ClientCodeServiceTest {
         Assertions.assertNotNull(clientCodeImdg.getSingleObjectByID(ID)); // verify test data
 
         //ACT
-        String jsonString = getJsonStringForUpdate(clientCodeDeleteRequest, ID);
+        String jsonString = TestUtils.getJsonStringForUPDATE(clientCodeDeleteRequest, ID);
 
-        addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_DELETE, PARTITION, 0, jsonString);
+        TestUtils.addRecordToKafka((MockConsumer) clientCodeService.getConsumer(), Consts.DESTINATION_CLIENT_CODE_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
 
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        TestUtils.waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
         ClientCode resultUpdate = clientCodeImdg.getSingleObjectByID(ID);
         Assertions.assertNull(resultUpdate);
     }

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/validation/CompanySymbolValidationConfig.java

@@ -38,7 +38,7 @@ public class CompanySymbolValidationConfig {
             ImdgValidationContext<CompanySymbolUpdateRequest> context = new ImdgValidationContext<>();
             context.setValidatedObject(companySymbolUpdateRequest);
             Consumer<String> addImdg = (s) -> context.addImdg(s, imdgForValidation.get(s));
-//            addImdg.accept(IMDGDistributedNames.Map_CompanySymbols);
+            addImdg.accept(IMDGDistributedNames.Map_CompanySymbols);
             addImdg.accept(IMDGDistributedNames.Map_CompanySymbolDictionary);
             addImdg.accept(IMDGDistributedNames.Map_Company);
             return new ValidatorImpl<>(context,

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/ClearingMemberCategoryServiceTest.java

@@ -1,16 +1,12 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.ClearingMemberCategory;
@@ -26,7 +22,6 @@ import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteReques
 import ru.spcex.clearing.platform.messaging.domain.cud.company.ClearingMemberCategoryNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.ClearingMemberCategoryUpdateRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.ClearingCategory;
@@ -36,8 +31,6 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 
 import javax.annotation.PostConstruct;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
@@ -66,10 +59,9 @@ class ClearingMemberCategoryServiceTest {
     private ImdgProvider hazelcastServiceTest;
     private Imdg<ClearingMemberCategory> memberCategoryImdg;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     private long COMPANY_ID;
     private Company TEST_COMPANY;
@@ -97,9 +89,6 @@ class ClearingMemberCategoryServiceTest {
         clearingCategoryDictionary = new ClearingCategoryDictionary();
         clearingCategoryDictionary.setCode(CLEARING_CATEGORY_DICT_CODE);
         clearingCategoryDictionaryImdg.insert(clearingCategoryDictionary);
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     @Test
@@ -120,7 +109,7 @@ class ClearingMemberCategoryServiceTest {
         addRecordToKafka((MockConsumer) clearingMemberCategoryService.getConsumer(), TOPIC_MEMBER_CATEGORY_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         ClearingMemberCategory resultNew = memberCategoryImdg.getSingleObjectBySQL(String.format("clearingMemberCategory = %s", clearingMemberCategory));
         predictableClearingMemberCategory.setId(resultNew.getId());
         MEMBER_CATEGORY_MATCHER.assertMatch(resultNew, predictableClearingMemberCategory);
@@ -156,7 +145,7 @@ class ClearingMemberCategoryServiceTest {
         addRecordToKafka((MockConsumer) clearingMemberCategoryService.getConsumer(), TOPIC_MEMBER_CATEGORY_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(id, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(id, mockProducer);
 
         ClearingMemberCategory resultUpdating = memberCategoryImdg.getSingleObjectByID(id);
         MEMBER_CATEGORY_MATCHER.assertMatch(resultUpdating, predictableClearingMemberCategory);
@@ -184,7 +173,7 @@ class ClearingMemberCategoryServiceTest {
         addRecordToKafka((MockConsumer) clearingMemberCategoryService.getConsumer(), TOPIC_MEMBER_CATEGORY_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(id, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(id, mockProducer);
 
         ClearingMemberCategory resultDeleting = memberCategoryImdg.getSingleObjectBySQL(String.format("clearingMemberCategory = %s", clearingMemberCategory));
         Assertions.assertNull(resultDeleting);

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/CompanyInfoServiceTest.java

@@ -1,15 +1,11 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -25,7 +21,6 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.CompanyInfoUpdateRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.WorkflowStatus;
@@ -34,8 +29,6 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 
 import javax.annotation.PostConstruct;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
@@ -70,10 +63,9 @@ class CompanyInfoServiceTest {
     private ImdgProvider hazelcastServiceTest;
     private Imdg<Company> companyImdg;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     @PostConstruct
     private void init() {
@@ -88,9 +80,6 @@ class CompanyInfoServiceTest {
         putToDictionary(IMDGDistributedNames.Map_AllowedDictionary, new AllowedDictionary(), "ALWD");
         putToDictionary(IMDGDistributedNames.Map_LegalKindDictionary, new LegalKindDictionary(), "JURD");
         putToDictionary(IMDGDistributedNames.Map_OrganizationTypeDictionary, new OrganizationTypeDictionary(), "NCRD");
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     private <D extends AbstractDictionary /*SpcexObjectBase & Dictionary*/> void putToDictionary(String mapName, D object, String code) {
@@ -172,7 +161,7 @@ class CompanyInfoServiceTest {
         addRecordToKafka((MockConsumer) companyInfoService.getConsumer(), TOPIC_COMPANY_INFO_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         Company resultUpdating = companyImdg.getSingleObjectByID(ID);
         COMPANY_INFO_MATCHER.assertMatch(resultUpdating.getProfile(), predictableCompanyInfo);

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/CompanyServiceTest.java

@@ -1,16 +1,12 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.ClearingMemberCategory;
@@ -30,7 +26,6 @@ import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.CompanyNewRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.CompanySymbol;
@@ -42,8 +37,6 @@ import javax.annotation.PostConstruct;
 import java.util.Arrays;
 import java.util.Map;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
@@ -78,10 +71,9 @@ class CompanyServiceTest {
     private Imdg<CompanySymbols> companySymbolsImdg;
     private Imdg<WorkflowStatusDictionary> workflowStatusDictionaryImdg;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     @PostConstruct
     private void init() {
@@ -115,9 +107,6 @@ class CompanyServiceTest {
             cioSymbol.setName(CompanySymbol.CIO.getKey());
             companySymbolDictionaryImdg.insert(cioSymbol);
         }
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     /**
@@ -143,7 +132,7 @@ class CompanyServiceTest {
         addRecordToKafka((MockConsumer) companyService.getConsumer(), TOPIC_COMPANY_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         //Company resultDeleting = companyImdg.getSingleObjectByID(ID);
         Company resultDeleting = companyImdg.getSingleObjectBySQL("id=" + ID);
@@ -186,7 +175,7 @@ class CompanyServiceTest {
         addRecordToKafka((MockConsumer) companyService.getConsumer(), DESTINATION_COMPANY_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         Company resultNew = companyImdg.getSingleObjectByFieldValues(Map.of("ShortName", "ClrIPO"));
         Assertions.assertNotNull(resultNew);
@@ -221,7 +210,6 @@ class CompanyServiceTest {
         //ASSERT
         waitingWhenTryAddRecordAndCheckError(ID,
                 mockProducer,
-                producerRecord,
                 String.valueOf(CompanyErrors.CompanyWithCompanySymbolAlreadyExist.getId()),
                 Arrays.asList("CIO", "test_value"));
 

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/CompanySymbolServiceTest.java

@@ -1,15 +1,11 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -25,7 +21,6 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.CompanySymbolUpdateRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.CompanySymbol;
@@ -34,8 +29,6 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 
 import javax.annotation.PostConstruct;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
@@ -69,10 +62,9 @@ class CompanySymbolServiceTest {
     private ImdgProvider hazelcastServiceTest;
     private Imdg<CompanySymbols> companySymbolsImdg;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     @PostConstruct
     private void init() {
@@ -102,10 +94,6 @@ class CompanySymbolServiceTest {
             company.setWorkflowStatus("ACTV");
             companyImdg.insert(company);
         }
-
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     /**
@@ -145,7 +133,7 @@ class CompanySymbolServiceTest {
         addRecordToKafka((MockConsumer) companySymbolService.getConsumer(), TOPIC_COMPANY_SYMBOL_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         CompanySymbols resultUpdating = companySymbolsImdg.getSingleObjectBySQL(String.format("companySymbolValue = %s", companySymbolValue));
         COMPANY_SYMBOL_MATCHER.assertMatch(resultUpdating, predictableCompanySymbols);

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/ContactServiceTest.java

@@ -1,15 +1,11 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -24,7 +20,6 @@ import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.ContactNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.ContactUpdateRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.ContactTypes;
@@ -34,8 +29,6 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 
 import javax.annotation.PostConstruct;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
@@ -64,10 +57,9 @@ class ContactServiceTest {
     private ImdgProvider hazelcastServiceTest;
     private Imdg<Contact> contactImdg;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     private Company TEST_COMPANY;
     private Long COMPANY_ID;
@@ -89,9 +81,6 @@ class ContactServiceTest {
                 IMDGDistributedNames.Map_ContactTypeDictionary, ContactTypeDictionary.class
         );
         contactTypeDictionaryImdg.insert(contactTypeDictionary);
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
     @Test
     void contactNew() throws InterruptedException {
@@ -111,7 +100,7 @@ class ContactServiceTest {
         addRecordToKafka((MockConsumer) contactService.getConsumer(), TOPIC_CONTACT_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         Contact resultNew = contactImdg.getAllValues().iterator().next();
         predictableContact.setId(resultNew.getId());
@@ -152,7 +141,7 @@ class ContactServiceTest {
         addRecordToKafka((MockConsumer) contactService.getConsumer(), TOPIC_CONTACT_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(existContactID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(existContactID, mockProducer);
 
         Contact resultUpdating = contactImdg.getSingleObjectByID(existContactID);
         CONTACT_MATCHER.assertMatch(resultUpdating, predictableContact);

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/ProfileDocumentServiceTest.java

@@ -1,16 +1,12 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -26,7 +22,6 @@ import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteReques
 import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentUpdateRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.DocumentTypes;
@@ -38,8 +33,6 @@ import javax.annotation.PostConstruct;
 import java.time.LocalDate;
 import java.time.temporal.ChronoUnit;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
@@ -71,10 +64,9 @@ class ProfileDocumentServiceTest {
     private Imdg<ProfileDocument> profileDocumentMap;
     private Imdg<Company> companyMap;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     private long COMPANY_ID;
     private Company TEST_COMPANY;
@@ -111,9 +103,6 @@ class ProfileDocumentServiceTest {
         documentTypeDictionary1.setCode(NEW_TEST_DOCUMENT_TYPE.getKey());
         documentTypeDictionary1.setName("new_test_name");
         documentTypeDictionaryImdg.insert(documentTypeDictionary1);
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     /**
@@ -159,7 +148,7 @@ class ProfileDocumentServiceTest {
         addRecordToKafka((MockConsumer) profileDocumentService.getConsumer(), TOPIC_DESTINATION_PROFILE_DOCUMENT_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         ProfileDocument resultNew = profileDocumentMap.getSingleObjectBySQL(String.format("companyId = %d", COMPANY_ID));
         predictableProfileDocument.setId(resultNew.getId());
         PROFILE_DOCUMENT_MATCHER.assertMatch(resultNew, predictableProfileDocument);
@@ -220,7 +209,7 @@ class ProfileDocumentServiceTest {
         addRecordToKafka((MockConsumer) profileDocumentService.getConsumer(), TOPIC_DESTINATION_PROFILE_DOCUMENT_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         ProfileDocument resultUpdate = profileDocumentMap.getSingleObjectBySQL(String.format("companyId = %d", COMPANY_ID));
 //        predictableProfileDocument.setId(resultUpdate.getId());
         PROFILE_DOCUMENT_MATCHER.assertMatch(resultUpdate, predictableProfileDocument);
@@ -255,7 +244,7 @@ class ProfileDocumentServiceTest {
         addRecordToKafka((MockConsumer) profileDocumentService.getConsumer(), TOPIC_DESTINATION_PROFILE_DOCUMENT_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         ProfileDocument resultUpdate = profileDocumentMap.getSingleObjectByID(PROFILE_DOCUMENT_ID);
         Assertions.assertNull(resultUpdate);
     }

clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/RelationServiceTest.java

@@ -1,15 +1,11 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -25,7 +21,6 @@ import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteReques
 import ru.spcex.clearing.platform.messaging.domain.cud.company.RelationNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.company.RelationUpdateRequest;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.ClearingCategory;
@@ -35,9 +30,7 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 
 import javax.annotation.PostConstruct;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
@@ -66,10 +59,9 @@ class RelationServiceTest {
     private Imdg<Relation> relationMap;
     private Imdg<Company> companyMap;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     private long COMPANY_ID;
 
@@ -109,15 +101,13 @@ class RelationServiceTest {
         clearingCategoryDictionary.setCode(ClearingCategory.I.getKey());
         clearingCategoryDictionary.setName("I");
         clearingCategoryDictionaryImdg.insert(clearingCategoryDictionary);
-
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
 
     @Test
     void newRelation() {
         //ARRANGE
+        clearAllInImdg(relationMap);
         RelationNewRequest relationNewRequest = new RelationNewRequest();
         relationNewRequest.setCompanyId(COMPANY_ID);
         relationNewRequest.setServiceStatus("ACTV");
@@ -138,7 +128,7 @@ class RelationServiceTest {
         addRecordToKafka((MockConsumer) relationService.getConsumer(), Consts.DESTINATION_RELATION_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         Relation resultNew = relationMap.getSingleObjectBySQL("serviceStatus=ACTV"); // String.format("supplierId = %d", COMPANY_ID));
         predictableRelation.setId(resultNew.getId());
         RELATION_MATCHER.assertMatch(resultNew, predictableRelation);
@@ -177,7 +167,7 @@ class RelationServiceTest {
         addRecordToKafka((MockConsumer) relationService.getConsumer(), Consts.DESTINATION_RELATION_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         Relation resultUpdate = relationMap.getSingleObjectBySQL(String.format("id = %d", RELATION_ID));
 //        predictableProfileDocument.setId(resultUpdate.getId());
         RELATION_MATCHER.assertMatch(resultUpdate, predictableRelation);
@@ -206,7 +196,7 @@ class RelationServiceTest {
         addRecordToKafka((MockConsumer) relationService.getConsumer(), Consts.DESTINATION_RELATION_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         Relation resultUpdate = relationMap.getSingleObjectByID(RELATION_ID);
 //        Assertions.assertNull(resultUpdate);
         assertEquals(WorkflowStatus.Blocked.getKey(), resultUpdate.getServiceStatus());

clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/AbstractServiceTest.java

@@ -1,13 +1,9 @@
 package ru.spcex.clearing.scheduler;
 
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -27,7 +23,6 @@ import ru.spcex.clearing.scheduler.config.validation.PlannerValidationConfig;
 import ru.spcex.clearing.scheduler.config.validation.ValidationConfig;
 import ru.spcex.clearing.scheduler.service.*;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.enumeration.*;
@@ -37,8 +32,6 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 import java.time.LocalDate;
 import java.time.LocalTime;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
@@ -74,10 +67,10 @@ public abstract class AbstractServiceTest {
     protected long testSecurityId = currentID.getAndIncrement();
     protected String sessionType = "IPOT";
 
-    @Captor
+    @Autowired
-    protected ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @MockBean
+    protected Producer<String, Object> mockProducer;
-    protected MockProducer<String, Object> mockProducer;
+
     @Autowired
     @Qualifier("hazelcastServiceTest")
     protected ImdgProvider imdgProvider;
@@ -133,9 +126,6 @@ public abstract class AbstractServiceTest {
         securityImdg.insert(security);
 
         clearingCalendarImdg.insert(getClearingCalendar());
-
-        TestUtils.FutureRecordMetadata future = spy(new TestUtils.FutureRecordMetadata());
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 
     protected PlannerTemplate getPlannerTemplate(long companyId) {

clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/service/ClearingCalendarServiceTest.java

@@ -75,7 +75,7 @@ class ClearingCalendarServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) clearingCalendarService.getConsumer(), TOPIC_CLEARING_CALENDAR_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         ClearingCalendar clearingCalendarRes = clearingCalendarImdg.getSingleObjectBySQL(String.format("companyId = %s", clearingCalendar.getCompanyId()));
         clearingCalendar.setId(clearingCalendarRes.getId());
@@ -113,7 +113,7 @@ class ClearingCalendarServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) clearingCalendarService.getConsumer(), TOPIC_CLEARING_CALENDAR_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         ClearingCalendar clearingCalendarRes = clearingCalendarImdg.getSingleObjectBySQL(String.format("companyId = %s", clearingCalendar.getCompanyId()));
         clearingCalendar.setId(clearingCalendarRes.getId());
@@ -148,7 +148,7 @@ class ClearingCalendarServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) clearingCalendarService.getConsumer(), TOPIC_CLEARING_CALENDAR_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(id, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(id, mockProducer);
 
         ClearingCalendar plannerTemplateRes = clearingCalendarImdg.getSingleObjectBySQL(String.format("companyId = %s", clearingCalendar.getCompanyId()));
         assertNull(plannerTemplateRes);

clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/service/LauncherServiceTest.java

@@ -1,8 +1,10 @@
 package ru.spcex.clearing.scheduler.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,6 +26,7 @@ import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparato
 import static ru.spcex.clearing.test.TestUtils.*;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.defaultAdminId;
+import static ru.spcex.clearing.test.config.KafkaTestConfig.getCaptor;
 import static ru.spcex.platform.enumeration.Task.accountBlock;
 
 class LauncherServiceTest extends AbstractServiceTest {
@@ -76,11 +79,12 @@ class LauncherServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) launcherService.getConsumer(), TOPIC_LAUNCHER_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
+        ArgumentCaptor<ProducerRecord> captor = getCaptor(mockProducer);
         verify(mockProducer, timeout(30_000L).times(1))
-                .send(producerRecord.capture());
+                .send(captor.capture());
 
-        BaseRequest<Object> baseRequestResult = (BaseRequest<Object>) producerRecord.getValue().value();
+        BaseRequest<Object> baseRequestResult = (BaseRequest<Object>) captor.getValue().value();
-        assertEquals("launcher-" + launcher.getTask(), producerRecord.getValue().topic());
+        assertEquals("launcher-" + launcher.getTask(), captor.getValue().topic());
         BASE_REQUEST_MATCHER.assertMatch(baseRequestResult, predictableBaseRequest);
 
 

clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/service/PlannerServiceTest.java

@@ -69,7 +69,7 @@ class PlannerServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) plannerService.getConsumer(), TOPIC_PLANNER_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         Planner plannerReq = plannerImdg.getSingleObjectBySQL(String.format("companyId = %s", planner.getCompanyId()));
         planner.setId(plannerReq.getId());
@@ -105,7 +105,7 @@ class PlannerServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) plannerService.getConsumer(), TOPIC_PLANNER_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(planner.getId(), mockProducer, producerRecord);
+        waitingSendAndCheckRecord(planner.getId(), mockProducer);
 
         Planner plannerReq = plannerImdg.getSingleObjectBySQL(String.format("companyId = %s", planner.getCompanyId()));
         planner.setId(plannerReq.getId());
@@ -137,7 +137,7 @@ class PlannerServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) plannerService.getConsumer(), TOPIC_PLANNER_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(plannerId, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(plannerId, mockProducer);
 
         Planner plannerReq = plannerImdg.getSingleObjectBySQL(String.format("companyId = %s", planner.getCompanyId()));
         assertNull(plannerReq);

clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/service/PlannerTemplateServiceTest.java

@@ -61,7 +61,7 @@ class PlannerTemplateServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) plannerTemplateService.getConsumer(), TOPIC_PLANNER_TEMPLATE_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         PlannerTemplate plannerTemplateRes = plannerTemplateImdg.getSingleObjectBySQL(String.format("companyId = %s", plannerTemplate.getCompanyId()));
         plannerTemplate.setId(plannerTemplateRes.getId());
@@ -96,7 +96,7 @@ class PlannerTemplateServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) plannerTemplateService.getConsumer(), TOPIC_PLANNER_TEMPLATE_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         PlannerTemplate plannerTemplateRes = plannerTemplateImdg.getSingleObjectBySQL(String.format("companyId = %s", plannerTemplate.getCompanyId()));
         plannerTemplate.setId(plannerTemplateRes.getId());
@@ -127,7 +127,7 @@ class PlannerTemplateServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) plannerTemplateService.getConsumer(), TOPIC_PLANNER_TEMPLATE_DELETE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(id, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(id, mockProducer);
 
         PlannerTemplate plannerTemplateRes = plannerTemplateImdg.getSingleObjectBySQL(String.format("companyId = %s", plannerTemplate.getCompanyId()));
         assertNull(plannerTemplateRes);

clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/service/TaskManagerTest.java

@@ -1,6 +1,8 @@
 package ru.spcex.clearing.scheduler.service;
 
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.springframework.beans.factory.annotation.Autowired;
 import ru.clearing.classes.statics.data.scheduler.PlannerAllToday;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
@@ -18,6 +20,7 @@ import static ru.spcex.clearing.scheduler.config.PlannerQueueConfig.addToPlanner
 import static ru.spcex.clearing.scheduler.service.TaskManager.systemId;
 import static ru.spcex.clearing.test.TestUtils.BASE_REQUEST_MATCHER;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
+import static ru.spcex.clearing.test.config.KafkaTestConfig.getCaptor;
 import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey;
 
 class TaskManagerTest extends AbstractServiceTest {
@@ -25,6 +28,9 @@ class TaskManagerTest extends AbstractServiceTest {
     @Autowired
     LauncherSender launcherSender;
 
+    @Autowired
+    private LauncherService launcherService;
+
     @PostConstruct
     public void init() {
         super.init();
@@ -95,8 +101,9 @@ class TaskManagerTest extends AbstractServiceTest {
     public void waitingWhenAddedLauncherCommandRequestAndCheckIt(Task toTaskQueue, Long userId) {
         BaseRequest<Object> predictableBaseRequest = launcherSender.makeCmdRequest(toTaskQueue, userId);
 
+        ArgumentCaptor<ProducerRecord> producerRecord = getCaptor(mockProducer);
         //waiting for kafka producer send message (finale event)
-        verify(mockProducer, timeout(60_000L).times(1))
+        verify(mockProducer, timeout(30_000L).times(1))
                 .send(producerRecord.capture());
 
         BaseRequest<Object> baseRequestResult = (BaseRequest<Object>) producerRecord.getValue().value();

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/AbstractServiceTest.java

@@ -1,13 +1,9 @@
 package ru.spcex.clearing.securities.service;
 
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.company.Company;
@@ -24,7 +20,6 @@ import ru.spcex.clearing.securities.config.ValidationConfig;
 import ru.spcex.clearing.securities.service.cud.*;
 import ru.spcex.clearing.securities.validation.ValidationProvider;
 import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.platform.imdg.api.Imdg;
@@ -32,8 +27,6 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
 
 import java.util.concurrent.atomic.AtomicLong;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
 
@@ -75,10 +68,9 @@ public abstract class AbstractServiceTest {
     public static String termType = "TermType";
     public static String currencyCode = "RUB";
 
-    @Captor
+    @Autowired
-    protected ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @MockBean
+    protected Producer<String, Object> mockProducer;
-    protected MockProducer<String, Object> mockProducer;
     @Autowired
     @Qualifier("hazelcastServiceTest")
     protected ImdgProvider imdgProvider;
@@ -127,7 +119,5 @@ public abstract class AbstractServiceTest {
         Company company = new Company();
         company.setId(issuerId);
         companyImdg.insert(company);
-        TestUtils.FutureRecordMetadata future = spy(TestUtils.FutureRecordMetadata.class);
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 }

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/CouponPeriodServiceTest.java

@@ -63,7 +63,7 @@ public class CouponPeriodServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) couponPeriodService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         CouponPeriod result = couponPeriodImdg.getSingleObjectBySQL(String.format("securityId = %s", currencyPrediction.getSecurityId()));
         currencyPrediction.setId(result.getId());
@@ -112,7 +112,7 @@ public class CouponPeriodServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) couponPeriodService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(id, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(id, mockProducer);
 
         CouponPeriod result = couponPeriodImdg.getSingleObjectBySQL(String.format("securityId = %s", predictableCoupon.getSecurityId()));
         predictableCoupon.setId(result.getId());

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/CurrencyServiceTest.java

@@ -54,7 +54,7 @@ public class CurrencyServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) currencyService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         Currency result = currencyImdg.getSingleObjectBySQL(String.format("countryCode = %s", currencyPrediction.getCountryCode()));
         currencyPrediction.setId(result.getId());
@@ -87,7 +87,7 @@ public class CurrencyServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) currencyService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         Currency result = currencyImdg.getSingleObjectBySQL(String.format("countryCode = %s", currencyPrediction.getCountryCode()));
         currencyPrediction.setId(result.getId());

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/EquitySecurityServiceTest.java

@@ -54,7 +54,7 @@ public class EquitySecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) equitySecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         EquitySecurity equityResult = equitySecurityImdg.getSingleObjectBySQL(String.format("fullName = %s", equityPrediction.getFullName()));
         equityPrediction.setId(equityResult.getId());
@@ -94,7 +94,7 @@ public class EquitySecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) equitySecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         EquitySecurity equityResult = equitySecurityImdg.getSingleObjectBySQL(String.format("fullName = %s", equityPrediction.getFullName()));
         equityPrediction.setId(equityResult.getId());
@@ -136,7 +136,7 @@ public class EquitySecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) equitySecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         EquitySecurity equityResult = equitySecurityImdg.getSingleObjectBySQL(String.format("fullName = %s", equityPrediction.getFullName()));
         equityPrediction.setId(equityResult.getId());

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/FixedIncomeCashFlowServiceTest.java

@@ -1,15 +1,11 @@
 package ru.spcex.clearing.securities.service;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.instrument.issue.FixedIncomeCashFlow;
@@ -59,10 +55,9 @@ class FixedIncomeCashFlowServiceTest {
     private ImdgProvider hazelcastServiceTest;
     private Imdg<FixedIncomeCashFlow> fixedIncomeCashFlowImdg;
 
-    @Captor
+    @Autowired
-    private ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("mockProducer")
-    @SpyBean
+    protected Producer<String, Object> mockProducer;
-    private MockProducer<String, Object> mockProducer;
 
     private final String SECURITY_SYMBOL_STR = "777";
     private final Long SECURITY_SYMBOL_LONG = 777L;
@@ -83,8 +78,9 @@ class FixedIncomeCashFlowServiceTest {
     }
 
     @Test
-    void fixedIncomeCashFlowNewTest() throws InterruptedException {
+    void fixedIncomeCashFlowNewTest() {
         //ARRANGE
+        clearAllInImdg(fixedIncomeCashFlowImdg);
         FixedIncomeCashFlowNewRequest newRequest = new FixedIncomeCashFlowNewRequest();
         newRequest.setSecuritySymbol(SECURITY_SYMBOL_STR);
         newRequest.setNominalValue(TEST_BIG_DECIMAL);
@@ -105,14 +101,14 @@ class FixedIncomeCashFlowServiceTest {
         addRecordToKafka((MockConsumer) fixedIncomeCashFlowService.getConsumer(), TOPIC_FIXED_INCOME_CASH_FLOW_NEW, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
         FixedIncomeCashFlow resultNew = fixedIncomeCashFlowImdg.getSingleObjectBySQL(String.format("securityId = %s", SECURITY_SYMBOL_STR));
         predictableFixedIncomeCashFlow.setId(resultNew.getId());
         FIXED_INCOME_CASH_FLOW_MATCHER.assertMatch(resultNew, predictableFixedIncomeCashFlow);
     }
 
     @Test
-    void fixedIncomeCashFlowUpdateTest() throws InterruptedException {
+    void fixedIncomeCashFlowUpdateTest() {
         //ARRANGE
         FixedIncomeCashFlow existFixedIncomeCashFlow = new FixedIncomeCashFlow();
         existFixedIncomeCashFlow.setSecurityId(SECURITY_SYMBOL_LONG);
@@ -143,7 +139,7 @@ class FixedIncomeCashFlowServiceTest {
         addRecordToKafka((MockConsumer) fixedIncomeCashFlowService.getConsumer(), TOPIC_FIXED_INCOME_CASH_FLOW_UPDATE, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(id, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(id, mockProducer);
         FixedIncomeCashFlow resultUpdate = fixedIncomeCashFlowImdg.getSingleObjectBySQL(String.format("securityId = %s", SECURITY_SYMBOL_STR));
         predictableFixedIncomeCashFlow.setId(id);
         FIXED_INCOME_CASH_FLOW_MATCHER.assertMatch(resultUpdate, predictableFixedIncomeCashFlow);

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/FixedIncomeSecurityServiceTest.java

@@ -55,7 +55,7 @@ public class FixedIncomeSecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) fixedIncomeSecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         FixedIncomeSecurity equityResult = fixedIncomeSecurityImdg.getSingleObjectBySQL(String.format("fullName = %s", fixedIncomePrediction.getFullName()));
         fixedIncomePrediction.setId(equityResult.getId());
@@ -96,7 +96,7 @@ public class FixedIncomeSecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) fixedIncomeSecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         FixedIncomeSecurity equityResult = fixedIncomeSecurityImdg.getSingleObjectBySQL(String.format("fullName = %s", fixedIncomePrediction.getFullName()));
         fixedIncomePrediction.setId(equityResult.getId());
@@ -138,7 +138,7 @@ public class FixedIncomeSecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) fixedIncomeSecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         FixedIncomeSecurity equityResult = fixedIncomeSecurityImdg.getSingleObjectBySQL(String.format("fullName = %s", fixedIncomePrediction.getFullName()));
         fixedIncomePrediction.setId(equityResult.getId());

clearing-parent/securities-service/src/test/java/ru/spcex/clearing/securities/service/MoneyMarketSecurityServiceTest.java

@@ -75,7 +75,7 @@ public class MoneyMarketSecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) moneyMarketSecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         MoneyMarketSecurity moneyMarketSecurityResult = moneyMarketSecurityMap.getSingleObjectBySQL(String.format("fullName = %s", moneyMarketSecurityFactory.getFullName()));
         moneyMarketSecurityPrediction.setId(moneyMarketSecurityResult.getId());
@@ -118,7 +118,7 @@ public class MoneyMarketSecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) moneyMarketSecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         MoneyMarketSecurity moneyMarketSecurityResult = moneyMarketSecurityMap.getSingleObjectBySQL(String.format("fullName = %s", moneyMarketSecurityFactory.getFullName()));
         moneyMarketSecurityPrediction.setId(moneyMarketSecurityResult.getId());
@@ -167,7 +167,7 @@ public class MoneyMarketSecurityServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) moneyMarketSecurityService.getConsumer(), TOPIC, PARTITION, 0, jsonString);
 
         //ASSERT
-        waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
+        waitingSendAndCheckRecord(ID, mockProducer);
 
         MoneyMarketSecurity moneyMarketSecurityResult = moneyMarketSecurityMap.getSingleObjectBySQL(String.format("fullName = %s", moneyMarketSecurityFactory.getFullName()));
         moneyMarketSecurityPrediction.setId(moneyMarketSecurityResult.getId());

clearing-parent/test-clearing/pom.xml

@@ -68,5 +68,9 @@
             <groupId>ru.spcex.platform</groupId>
             <artifactId>platform-enum</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-test</artifactId>
+        </dependency>
     </dependencies>
 </project>

clearing-parent/test-clearing/src/main/java/ru/spcex/clearing/test/TestUtils.java

@@ -4,11 +4,14 @@ import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
 import ru.spcex.clearing.platform.messaging.domain.ActionType;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.domain.Consts;
@@ -17,6 +20,7 @@ import ru.spcex.clearing.platform.messaging.service.Status;
 import ru.spcex.platform.classes.base.SpcexObjectBase;
 import ru.spcex.platform.imdg.api.Imdg;
 
+import java.lang.reflect.Field;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -27,17 +31,50 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.*;
-import static org.mockito.Mockito.verify;
 import static ru.spcex.clearing.platform.messaging.service.Status.Success;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.defaultAdminId;
+import static ru.spcex.clearing.test.config.KafkaTestConfig.getCaptor;
 
 public class TestUtils {
     public static final MatcherFactory.Matcher<BaseRequest<Object>> BASE_REQUEST_MATCHER = usingIgnoringFieldsComparator();
     private static final ObjectMapper objectMapper = new ObjectMapper();
 
-    public static void waitingWhenAddedRecordAndCheckIt(Long id, MockProducer mockProducer, ArgumentCaptor<ProducerRecord> producerRecord) {
+    public static void waitingSendAndCheckRecord(Long id, Producer mockProducer, ArgumentCaptor<ProducerRecord> producerRecord) {
+        //waiting for kafka send message (finale event)
+        verify(mockProducer, timeout(30_000L).times(1))
+                .send(producerRecord.capture());
+        checkRecord(id, producerRecord);
+    }
+
+    public static void waitingSendAndCheckRecord(Long id, Producer mockProducer) {
+        ArgumentCaptor<ProducerRecord> producerRecord = getCaptor(mockProducer);
+        //waiting for kafka send message (finale event)
+        verify(mockProducer, timeout(30_000L).times(1))
+                .send(producerRecord.capture());
+        checkRecord(id, producerRecord);
+    }
+
+    public static void waitingSendAndCheckRecord(Long id, KafkaTemplate<String, Object> kafkaTemplate) {
+        ArgumentCaptor<ProducerRecord> producerRecord = getCaptor(kafkaTemplate);
+        //waiting for kafka send message (finale event)
+        verify(kafkaTemplate, timeout(30_000L).times(1))
+                .send(producerRecord.capture());
+        checkRecord(id, producerRecord);
+    }
+
+    public static void waitingSendAndReceiveAndCheckRecord(Long id, KafkaTemplate<String, Object> kafkaTemplate) {
+        ReplyingKafkaTemplate<String, Object, Object> rplKafkaTemplate = (ReplyingKafkaTemplate<String, Object, Object>) kafkaTemplate;
+        ArgumentCaptor<ProducerRecord> producerRecord = getCaptor(kafkaTemplate);
+
+        //waiting for kafka send message (finale event)
+        verify(rplKafkaTemplate, timeout(30_000L).times(1))
+                .sendAndReceive(producerRecord.capture());
+        checkRecord(id, producerRecord);
+    }
+
+    public static void checkRecord(Long id, ArgumentCaptor<ProducerRecord> producerRecord) {
         BaseRequest<Object> predictableBaseRequest = new BaseRequest<>();
         predictableBaseRequest.setId(id);
         predictableBaseRequest.setActionType(ActionType.SYSTEM);
@@ -46,20 +83,16 @@ public class TestUtils {
         requestInfoUpdate.setStatus(Success);
         predictableBaseRequest.setRequestPayload(requestInfoUpdate);
 
-        //waiting for kafka producer send message (finale event)
-        verify(mockProducer, timeout(30_000L).times(1))
-                .send(producerRecord.capture());
-
         BaseRequest<Object> baseRequestResult = (BaseRequest<Object>) producerRecord.getValue().value();
         assertEquals(Consts.REQUEST_INFO_UPDATE, producerRecord.getValue().topic());
         BASE_REQUEST_MATCHER.assertMatch(baseRequestResult, predictableBaseRequest);
     }
 
     public static void waitingWhenTryAddRecordAndCheckError(Long id,
-                                                            MockProducer mockProducer,
+                                                            Producer mockProducer,
-                                                            ArgumentCaptor<ProducerRecord> producerRecord,
                                                             String errorCode,
                                                             List<String> errorMessageArgs) {
+        ArgumentCaptor<ProducerRecord> producerRecord = getCaptor(mockProducer);
         BaseRequest<Object> predictableBaseRequest = new BaseRequest<>();
         predictableBaseRequest.setId(id);
         predictableBaseRequest.setActionType(ActionType.SYSTEM);
@@ -122,6 +155,21 @@ public class TestUtils {
         values.forEach(imdg::delete);
     }
 
+    public static <T, V> T getMockForFildObj(T from, V obj, String fildName) {
+        T mock = Mockito.mock((Class<T>) from.getClass(), withSettings()
+                .serializable()
+                .spiedInstance(from)
+                .defaultAnswer(CALLS_REAL_METHODS));
+        try {
+            Field dbServiceField = obj.getClass().getDeclaredField(fildName);
+            dbServiceField.setAccessible(true);
+            dbServiceField.set(obj, mock);
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            throw new RuntimeException(e);
+        }
+        return mock;
+    }
+
     public static class FutureRecordMetadata implements Future<RecordMetadata> {
         @Override
         public boolean cancel(boolean mayInterruptIfRunning) {

clearing-parent/test-clearing/src/main/java/ru/spcex/clearing/test/config/KafkaTestConfig.java

@@ -1,31 +1,135 @@
 package ru.spcex.clearing.test.config;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.mockito.ArgumentCaptor;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.boot.test.mock.mockito.MockReset;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Scope;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
+import org.springframework.kafka.requestreply.RequestReplyFuture;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.util.concurrent.ListenableFuture;
 import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgId;
 import ru.spcex.platform.imdg.api.ImdgProvider;
 
-@Configuration
+import java.util.HashMap;
-public class KafkaTestConfig {
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+import static org.mockito.Mockito.*;
+
+@Configuration
+@Import(ImdgTestConfig.class)
+public class KafkaTestConfig {
+    public final static Map<Producer<String, Object>, ArgumentCaptor<ProducerRecord>> producerCaptors = new HashMap<>();
+    public final static Map<KafkaTemplate<String, Object>, ArgumentCaptor<ProducerRecord>> templateCaptors = new HashMap<>();
+
+    //из-за очисткой перед каждым тестом(MockReset.withSettings(MockReset.AFTER) необходимо каждый раз обновлять doReturn
+    public static ArgumentCaptor<ProducerRecord> getCaptor(Producer<String, Object> mockProducer){
+        ArgumentCaptor<ProducerRecord> captor = producerCaptors.get(mockProducer);
+        setFuture(captor, mockProducer);
+        return captor;
+    }
+
+    public static ArgumentCaptor<ProducerRecord> getCaptor(KafkaTemplate<String, Object> kafkaTemplate){
+        ArgumentCaptor<ProducerRecord> captor = templateCaptors.get(kafkaTemplate);
+        setFuture(captor, (ReplyingKafkaTemplate<String, Object, Object>) kafkaTemplate);
+        return captor;
+    }
+
+    public static void setFuture(ArgumentCaptor<ProducerRecord> captor, Producer<String, Object> mockProducer){
+        TestUtils.FutureRecordMetadata future = spy(new TestUtils.FutureRecordMetadata());
+        doReturn(future).when(mockProducer).send(captor.capture());
+    }
+    public static void setFuture(ArgumentCaptor<ProducerRecord> recordArgumentCaptor, ReplyingKafkaTemplate<String, Object, Object> kafkaTemplate){
+
+        //sendToQueueWaitForAnswer
+        RequestReplyFuture<String, Object, Object> replyFuture = spy(RequestReplyFuture.class);
+
+        doReturn(replyFuture).when(kafkaTemplate).sendAndReceive(recordArgumentCaptor.capture());
+
+        ConsumerRecord<String, Object> consumerRecord = mock(ConsumerRecord.class);
+        try {
+            doReturn(consumerRecord).when(replyFuture).get(10, TimeUnit.SECONDS);
+        } catch (InterruptedException | ExecutionException | TimeoutException | ClassCastException e) {
+            if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        doReturn(null).when(consumerRecord).value();
+
+        //sendRequestToQueue
+        ListenableFuture<SendResult<String, Object>> send = mock(ListenableFuture.class);
+        doReturn(send).when(kafkaTemplate).send(recordArgumentCaptor.capture());
+        try {
+            doReturn(null).when(send).get();
+        } catch (InterruptedException | ExecutionException e) {
+            if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+//    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+    @Bean("mockProducer")
+    public Producer<String, Object> kafkaProducer() {
+        Producer<String, Object> mockProducer = mock(MockProducer.class, MockReset.withSettings(MockReset.AFTER));
+        ArgumentCaptor<ProducerRecord> recordArgumentCaptor = ArgumentCaptor.forClass(ProducerRecord.class);
+        setFuture(recordArgumentCaptor, mockProducer);
+        producerCaptors.put(mockProducer, recordArgumentCaptor);
+        return mockProducer;
+    }
 
-    @Autowired
     @Bean
-    public KafkaSender kafkaSender(Producer<String, Object> kafkaProducer,
+    public KafkaSender kafkaSender(ImdgProvider imdgProvider, Producer<String, Object> mockProducer) {
-                                   ImdgProvider imdgProvider) {
         ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
         return KafkaSender
                 .setup()
-                .producer(kafkaProducer)
+                .producer(mockProducer)
+                .idGenerator(imdgIdGenerator::nextId)
+                .imdgProvider(s -> {
+                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
+                    return imdg::insert;
+                })
+                .build();
+    }
+
+    @Bean("kafkaTestTemplate")
+    public KafkaTemplate<String, Object> kafkaTemplate() {
+        ReplyingKafkaTemplate<String, Object, Object> kafkaTemplate = mock(ReplyingKafkaTemplate.class, MockReset.withSettings(MockReset.AFTER));
+        ArgumentCaptor<ProducerRecord> recordArgumentCaptor = ArgumentCaptor.forClass(ProducerRecord.class);
+        setFuture(recordArgumentCaptor, kafkaTemplate);
+        templateCaptors.put(kafkaTemplate, recordArgumentCaptor);
+        return kafkaTemplate;
+    }
+
+    @Autowired
+    @Bean
+    public Supplier<KafkaSender> kafkaSenderSupplier(KafkaTemplate<String, Object> kafkaTemplate,
+                                                     @Qualifier("hazelcastServiceTest") ImdgProvider imdgProvider) {
+        ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
+        return () -> KafkaSender
+                .setup()
+                .setKafkaTemplate(kafkaTemplate)
                 .idGenerator(imdgIdGenerator::nextId)
                 .imdgProvider(s -> {
                     Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);

clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java

@@ -40,7 +40,7 @@ public class TradeImporterService {
     private final Supplier<KafkaSender> kafka;
     private final IMessageResolver messageResolver;
 
-    @Value("${trade-importer.database.schema}")
+    @Value("${trade-importer.database.schema:SPVB_TS}")
     private String schema;
 
 

clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/AbstractServiceTest.java

@@ -1,20 +1,13 @@
 package ru.spcex.clearing.trade.importer;
 
-import org.apache.kafka.clients.producer.MockProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import ru.clearing.classes.statics.data.misc.STrades;
-import ru.clearing.classes.statics.data.scheduler.PlannerAllToday;
 import ru.spcex.clearing.imdg.IMDGDistributedNames;
-import ru.spcex.clearing.test.MatcherFactory;
-import ru.spcex.clearing.test.TestUtils;
 import ru.spcex.clearing.test.config.ImdgTestConfig;
 import ru.spcex.clearing.test.config.KafkaTestConfig;
 import ru.spcex.clearing.trade.importer.config.ErrorResolverConfig;
@@ -25,9 +18,6 @@ import ru.spcex.clearing.trade.importer.services.TradeImporterService;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgProvider;
 
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
-import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
 import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
 
@@ -41,14 +31,13 @@ import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProv
         ImdgTestConfig.class,
         KafkaTestConfig.class})
 public abstract class AbstractServiceTest {
-    protected static final MatcherFactory.Matcher<PlannerAllToday> PLANNER_ALL_TODAY_MATCHER = usingIgnoringFieldsComparator("created", "updated");
     protected static final long id = currentID.getAndIncrement();
     protected Imdg<STrades> sTradesImdg;
 
-    @Captor
+    @Autowired
-    protected ArgumentCaptor<ProducerRecord> producerRecord;
+    @Qualifier("kafkaTestTemplate")
-    @MockBean
+    protected KafkaTemplate<String, Object> kafkaTemplate;
-    protected MockProducer<String, Object> mockProducer;
+
     @Autowired
     @Qualifier("hazelcastServiceTest")
     protected ImdgProvider imdgProvider;
@@ -56,8 +45,5 @@ public abstract class AbstractServiceTest {
     protected void init() {
         waitAvailableImdgProviderAndAddAdminWithDefaultId();
         this.sTradesImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_STrades, STrades.class);
-
-        TestUtils.FutureRecordMetadata future = spy(new TestUtils.FutureRecordMetadata());
-        doReturn(future).when(mockProducer).send(producerRecord.capture());
     }
 }

clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/config/config/DbTestConnectionConfig.java

@@ -20,7 +20,7 @@ public class DbTestConnectionConfig {
     public SingleConnectionDataSource dataSource() {
         String login = "sa";
         String password = "Aa123456";
-        String dbUrl = "jdbc:sqlserver://localhost:1433;database=SPVB_TS;schema=dbo";
+        String dbUrl = "jdbc:sqlserver://10.200.200.144:1433;database=ni";
 
 
         SingleConnectionDataSource cpds = new SingleConnectionDataSource();

clearing-parent/trade-importer/src/test/java/ru/spcex/clearing/trade/importer/services/TradeImporterServiceTest.java

@@ -1,9 +1,11 @@
 package ru.spcex.clearing.trade.importer.services;
 
 import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.springframework.beans.factory.annotation.Autowired;
 import ru.clearing.classes.statics.data.misc.STrades;
 import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
@@ -22,6 +24,7 @@ import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparator;
 import static ru.spcex.clearing.test.TestUtils.*;
+import static ru.spcex.clearing.test.config.KafkaTestConfig.getCaptor;
 
 class TradeImporterServiceTest extends AbstractServiceTest {
     public static final MatcherFactory.Matcher<STrades> S_TRADES_MATCHER = usingIgnoringFieldsComparator();
@@ -63,8 +66,9 @@ class TradeImporterServiceTest extends AbstractServiceTest {
         addRecordToKafka((MockConsumer) launcherCommandReceiver.getConsumer(), Task.getOfTrades.topic(), 0, 1, getJsonStringForNew(new LauncherCommandRequest(),0));
 
         //waiting for kafka producer send message
-        verify(mockProducer, timeout(30_000L).times(1))
+        ArgumentCaptor<ProducerRecord> captor = getCaptor(kafkaTemplate);
-                .send(producerRecord.capture());
+        verify(kafkaTemplate, timeout(30_000L).times(1))
+                .send(captor.capture());
 
         STrades sTrades = tradeImporterService.getSTradesFromImdg(sTrade, sTradesImdg);
 

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java

@@ -89,7 +89,6 @@ public interface Consts {
     String DESTINATION_TRADING_CLEARING_REGISTRY_NEW = "trading-clearing-registry-new"; // todo check buplicate REGISTRY_NEW?
     String DESTINATION_TRADING_CLEARING_REGISTRY_AUTO_NEW = "trading-clearing-registry-auto-new";
     String DESTINATION_TRADING_CLEARING_REGISTRY_UPDATE = "trading-clearing-registry-update";
-    String DESTINATION_TRADING_CLEARING_REGISTRY_REPLY = "trading-clearing-registry-reply";
     String DESTINATION_TRADING_CLEARING_REGISTRY_BLOCK = "trading-clearing-registry-block";
 
     String DESTINATION_SDF08_NEW = "s-df-08-new";

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java

@@ -163,7 +163,7 @@ public class QueueConsumer implements AutoCloseable {
         });
     }
 
-    private void sendResponse(BaseRequest<?> o, Object response, Header correlationId) {
+    protected void sendResponse(BaseRequest<?> o, Object response, Header correlationId) {
         outputExecutor.submit(() -> {
             try {
                 Future<RecordMetadata> send;