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

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

@@ -8,6 +8,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
 import ru.spcex.clearing.account.config.settings.AccountServiceSettings;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
 import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
@@ -41,7 +42,7 @@ public class KafkaConfig {
                 .producer(kafkaProducer)
                 .idGenerator(imdgIdGenerator::nextId)
                 .imdgProvider(s -> {
-                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
+                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
                     return imdg::insert;
                 })
                 .build();

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

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.account.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -23,8 +24,8 @@ public class BankAccountService extends QueueConsumer implements InitializingBea
     private final Imdg<BankAccount> bankAccountMap;
 
     @Autowired
-    public BankAccountService(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
+    public BankAccountService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer, ImdgProvider imdgProvider) {
-        super(kafkaQueue);
+        super(kafkaQueue, kafkaProducer);
         this.bankAccountMap = imdgProvider.getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class);
     }
 

clearing-parent/backend-api/pom.xml

@@ -53,6 +53,10 @@
             <groupId>ru.spcex.clearing</groupId>
             <artifactId>classes</artifactId>
         </dependency>
+        <dependency>
+            <groupId>ru.spcex.clearing</groupId>
+            <artifactId>dictionary</artifactId>
+        </dependency>
     </dependencies>
 
     <dependencyManagement>
@@ -94,6 +98,37 @@
                     <finalName>${project.artifactId}</finalName>
                 </configuration>
             </plugin>
+
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>xml-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>transform</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <transformationSets>
+                        <transformationSet>
+                            <dir>src/data/meta</dir>
+                            <excludes>
+                                <exclude>src/data/meta/meta.server.html</exclude>
+                                <exclude>src/data/meta/meta.server.json</exclude>
+                                <exclude>src/data/meta/meta.server.xslt</exclude>
+                            </excludes>
+                            <stylesheet>src/data/meta/meta.server.xslt</stylesheet>
+                            <fileMappers>
+                                <fileMapper
+                                        implementation="org.codehaus.plexus.components.io.filemappers.FileExtensionMapper">
+                                    <targetExtension>.json</targetExtension>
+                                </fileMapper>
+                            </fileMappers>
+                        </transformationSet>
+                    </transformationSets>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java

@@ -1,5 +1,8 @@
 package ru.spcex.clearing.backendapi.config;
 
+import org.apache.kafka.clients.consumer.Consumer;
+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.springframework.beans.factory.annotation.Autowired;
@@ -7,6 +10,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
 import ru.spcex.clearing.backendapi.config.element.BackendApiSettings;
+import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
 import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 
 @Configuration
@@ -16,7 +20,7 @@ public class KafkaConfig {
     @Autowired
     @Bean
     public Producer<String, Object> createProducer(BackendApiSettings settings) {
-        return KafkaProducerFactory.producer(settings.getKafka());
+        return KafkaProducerFactory.producer(settings.getKafkaProducer());
     }
 
     @Profile("kafkaDisabled")
@@ -25,4 +29,17 @@ public class KafkaConfig {
         return new MockProducer<>(true, (topic, data) -> data != null ? data.getBytes() : new byte[0], (topic, data) -> data.toString().getBytes());
     }
 
+    @Profile("!kafkaDisabled")
+    @Autowired
+    @Bean
+    public Consumer<String, Object> createConsumer(BackendApiSettings settings) {
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
+    }
+
+    @Profile("kafkaDisabled")
+    @Bean
+    public Consumer<String, Object> createConsumerMock() {
+        return new MockConsumer<>(OffsetResetStrategy.NONE);
+    }
+
 }

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java

@@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.PropertySource;
 import org.springframework.stereotype.Component;
 import ru.spcex.clearing.backendapi.security.element.SecuritySettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
 import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
@@ -12,7 +13,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 @ConfigurationProperties("backend-api")
 public class BackendApiSettings {
     private HazelcastClientParams hazelcast;
-    private KafkaProducerSettings kafka;
+    private KafkaProducerSettings kafkaProducer;
+    private KafkaConsumerSettings kafkaConsumer;
     private SecuritySettings security;
     private String exampleSetting;
 
@@ -24,12 +26,20 @@ public class BackendApiSettings {
         this.hazelcast = hazelcast;
     }
 
-    public KafkaProducerSettings getKafka() {
+    public KafkaProducerSettings getKafkaProducer() {
-        return kafka;
+        return kafkaProducer;
     }
 
-    public void setKafka(KafkaProducerSettings kafka) {
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
-        this.kafka = kafka;
+        this.kafkaProducer = kafkaProducer;
+    }
+
+    public KafkaConsumerSettings getKafkaConsumer() {
+        return kafkaConsumer;
+    }
+
+    public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
+        this.kafkaConsumer = kafkaConsumer;
     }
 
     public String getExampleSetting() {

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/QueueExceptionHandler.java

@@ -39,7 +39,7 @@ public class QueueExceptionHandler extends ResponseEntityExceptionHandler {
     @ResponseStatus(HttpStatus.BAD_REQUEST)
     @ResponseBody
     public BasicSpcexResponse handleValidationException(ActionValidationException e) {
-        log.error("Default scoring exception handler: {}", ExceptionUtils.getStackTrace(e));
+        log.error("Default exception handler: {}", ExceptionUtils.getStackTrace(e));
         EnumMessage firstError = e.getErrors().iterator().next();
         BasicSpcexResponse errorResponse = new BasicSpcexResponse();
         errorResponse.setCode(firstError.getSubject().getId());
@@ -68,7 +68,7 @@ public class QueueExceptionHandler extends ResponseEntityExceptionHandler {
     @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
     @ResponseBody
     public BasicSpcexResponse handleDefault(Throwable e) {
-        log.error("Default scoring exception handler: {}", ExceptionUtils.getStackTrace(e));
+        log.error("Default exception handler: {}", ExceptionUtils.getStackTrace(e));
         BasicSpcexResponse errorResponse = new BasicSpcexResponse();
         errorResponse.setCode(-1);
         errorResponse.setMessage("internal server error");

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetAll.java

@@ -0,0 +1,49 @@
+package ru.spcex.clearing.backendapi.controller.response.entity.dictionary;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import ru.clearing.platform.dictionary.AbstractDictionary;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+@ApiModel(description = "Ответ при получении объектов словаря.")
+public class DictionaryBackendGetAll extends BasicSpcexResponse {
+
+    @JsonProperty
+    @ApiModelProperty(value = "Полезная нагрузка")
+    private DictionaryBackendPayload payload = new DictionaryBackendPayload();
+
+    public void fromEntity(Collection<AbstractDictionary> dictionaryValues) {
+        var payload = this.getPayload();
+        for (AbstractDictionary dictionaryValue : dictionaryValues) {
+            var singleItem = new DictionaryBackendGetFields();
+            singleItem.fromEntity(dictionaryValue);
+            payload.getItems().add(singleItem);
+        }
+    }
+
+
+    private static class DictionaryBackendPayload {
+        private List<DictionaryBackendGetFields> items = new ArrayList<>();
+
+        public List<DictionaryBackendGetFields> getItems() {
+            return items;
+        }
+
+        public void setItems(List<DictionaryBackendGetFields> items) {
+            this.items = items;
+        }
+    }
+
+    public DictionaryBackendPayload getPayload() {
+        return payload;
+    }
+
+    public void setPayload(DictionaryBackendPayload payload) {
+        this.payload = payload;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetFields.java

@@ -0,0 +1,48 @@
+package ru.spcex.clearing.backendapi.controller.response.entity.dictionary;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModelProperty;
+import ru.clearing.platform.dictionary.AbstractDictionary;
+
+public class DictionaryBackendGetFields {
+    @JsonProperty
+    @ApiModelProperty(value = "Идентификатор", example = "1234")
+    private Long id;
+    @JsonProperty
+    @ApiModelProperty(value = "", example = "")
+    private String code;
+    @JsonProperty
+    @ApiModelProperty(value = "Имя банка", example = "Сбербанк")
+    private String name;
+
+    public void fromEntity(AbstractDictionary dictionaryValue) {
+        this.id = dictionaryValue.getId();
+        this.code = dictionaryValue.getCode();
+        this.name = dictionaryValue.getName();
+    }
+
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetSingleValue.java

@@ -0,0 +1,30 @@
+package ru.spcex.clearing.backendapi.controller.response.entity.dictionary;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import ru.clearing.platform.dictionary.AbstractDictionary;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+
+@ApiModel(description = "Ответ при получении объекта BankAccount.")
+public class DictionaryBackendGetSingleValue extends BasicSpcexResponse {
+
+    @JsonProperty
+    @ApiModelProperty(value = "Поля объекта")
+    private DictionaryBackendGetFields payload;
+
+    public DictionaryBackendGetFields getPayload() {
+        return payload;
+    }
+
+    public void fromEntity(AbstractDictionary dictionaryValue) {
+        var payload = new DictionaryBackendGetFields();
+        this.setPayload(payload);
+        payload.fromEntity(dictionaryValue);
+    }
+
+
+    public void setPayload(DictionaryBackendGetFields payload) {
+        this.payload = payload;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/system/RequestInfoResponse.java

@@ -0,0 +1,51 @@
+package ru.spcex.clearing.backendapi.controller.response.system;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+import ru.spcex.clearing.platform.messaging.domain.json.serialize.EnumSerializer;
+import ru.spcex.clearing.platform.messaging.service.Status;
+
+
+/**
+ * Банковские реквизиты для перечисления денежных средств
+ * @ApiModel(description = "Ответ в результате отправки операции в топик Kafka.")
+ * @ApiModelProperty(value = "Информация о принятом запросе")
+ **/
+@ApiModel(description = "Ответ при получении статуса запроса.")
+public class RequestInfoResponse extends BasicSpcexResponse {
+
+    @JsonProperty
+    @ApiModelProperty(value = "Поля объекта")
+    private RequestInfoPayload payload;
+
+    public RequestInfoPayload getPayload() {
+        return payload;
+    }
+
+    public void setStatus(Status status) {
+        this.payload = new RequestInfoPayload();
+        payload.setStatus(status);
+    }
+
+    public void setPayload(RequestInfoPayload payload) {
+        this.payload = payload;
+    }
+
+    private static class RequestInfoPayload {
+        @JsonProperty
+        @JsonSerialize(using = EnumSerializer.class)
+        private Status status;
+
+        public Status getStatus() {
+            return status;
+        }
+
+        public void setStatus(Status status) {
+            this.status = status;
+        }
+    }
+
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/DictionaryController.java

@@ -0,0 +1,139 @@
+package ru.spcex.clearing.backendapi.controller.system;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.*;
+import ru.clearing.platform.dictionary.AbstractDictionary;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+import ru.spcex.clearing.backendapi.controller.response.entity.dictionary.DictionaryBackendGetAll;
+import ru.spcex.clearing.backendapi.controller.response.entity.dictionary.DictionaryBackendGetSingleValue;
+import ru.spcex.clearing.backendapi.errors.ActionValidationException;
+import ru.spcex.clearing.backendapi.errors.BackEndError;
+import ru.spcex.clearing.backendapi.errors.NotFound404Exception;
+import ru.spcex.platform.classes.base.SpcexObjectBase;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.utils.enumeration.EnumMessage;
+import ru.spcex.platform.utils.enumeration.IMessageResolver;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Controller
+@RequestMapping("/dictionaries")
+public class DictionaryController {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final Map<String, Imdg<?>> allDictionaryMaps;
+    private final ImdgProvider imdgProvider;
+
+    @Autowired
+    public DictionaryController(ImdgProvider imdgProvider) {
+        this.imdgProvider = imdgProvider;
+        this.allDictionaryMaps = new ConcurrentHashMap<>();
+    }
+
+    @ApiOperation(value = "get all dictionary value.")
+    @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = DictionaryBackendGetAll.class)})
+    @RequestMapping(value = "/{dictionary-name}", method = RequestMethod.GET)
+    @ResponseBody
+    public DictionaryBackendGetAll getAll(@ApiParam(value = "Название словаря", required = true, example = "moneyFlowSide")
+                                          @PathVariable("dictionary-name") String dictionaryName) {
+        Imdg<AbstractDictionary> dictionary = extractDictionaryImdgFromUrlParameter(dictionaryName, AbstractDictionary.class);
+        Collection<AbstractDictionary> allValues = dictionary.getAllValues();
+        DictionaryBackendGetAll response = new DictionaryBackendGetAll();
+        response.fromEntity(allValues);
+        return response;
+    }
+
+    @ApiOperation(value = "get dictionary value by id.")
+    @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = DictionaryBackendGetSingleValue.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)})
+    @RequestMapping(value = "/{dictionary-name}/{id}", method = RequestMethod.GET)
+    @ResponseBody
+    public DictionaryBackendGetSingleValue getById(@ApiParam(value = "Название словаря", required = true, example = "moneyFlowSide")
+                                                       @PathVariable("dictionary-name") String dictionaryName,
+                                                   @ApiParam(value = "Идентификатор объекта", required = true, example = "1234")
+                                             @PathVariable("id") Long id) {
+        Imdg<AbstractDictionary> dictionary = extractDictionaryImdgFromUrlParameter(dictionaryName, AbstractDictionary.class);
+        AbstractDictionary value = dictionary.getSingleObjectByID(id);
+        if (value == null) throw new NotFound404Exception(dictionaryName + " id='" + id + "'");
+        DictionaryBackendGetSingleValue response = new DictionaryBackendGetSingleValue();
+        response.fromEntity(value);
+        return response;
+    }
+
+    @ApiOperation(value = "get dictionary value by code.")
+    @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = DictionaryBackendGetSingleValue.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)})
+    @RequestMapping(value = "/{dictionary-name}/code/{code}", method = RequestMethod.GET)
+    @ResponseBody
+    public DictionaryBackendGetSingleValue getByCode(@ApiParam(value = "Название словаря", required = true, example = "moneyFlowSide")
+                                                       @PathVariable("dictionary-name") String dictionaryName,
+                                                   @ApiParam(value = "Код словаря", required = true, example = "1234")
+                                             @PathVariable("code") String code) {
+        Imdg<AbstractDictionary> dictionary = extractDictionaryImdgFromUrlParameter(dictionaryName, AbstractDictionary.class);
+        AbstractDictionary value = dictionary.getSingleObjectByFieldValues(Map.of("code", code));
+        if (value == null) throw new NotFound404Exception(dictionaryName + " code='" + code + "'");
+        DictionaryBackendGetSingleValue response = new DictionaryBackendGetSingleValue();
+        response.fromEntity(value);
+        return response;
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T extends SpcexObjectBase> Imdg<T> extractDictionaryImdgFromUrlParameter(String dictionaryNameFromUrl, Class<T> clazz) {
+        if (dictionaryNameFromUrl == null || dictionaryNameFromUrl.length() < 1) {
+            throw new ActionValidationException(BackEndError.ValidationError, "dictionary-name");
+        }
+
+        String actualDictionaryName = "Map_"
+                + dictionaryNameFromUrl.substring(0, 1).toUpperCase()
+                + dictionaryNameFromUrl.substring(1)
+                + "Dictionary";
+
+        Imdg<T> dictionary = (Imdg<T>) allDictionaryMaps.computeIfAbsent(actualDictionaryName, (mapName1) -> {
+            Imdg<T> potentialImdg = imdgProvider.getImdg(actualDictionaryName, clazz);
+            if (potentialImdg.size() == 0) {
+                log.warn("coudln't find dictionary {}", actualDictionaryName);
+                return null;
+            }
+            return potentialImdg;
+        });
+
+        if (dictionary == null) {
+            throw new ActionValidationException(BackEndError.DictionaryNotFound, dictionaryNameFromUrl);
+        }
+
+        return dictionary;
+    }
+
+    @Autowired
+    private IMessageResolver errorResolver;
+
+    @ResponseStatus(value = HttpStatus.BAD_REQUEST)
+    @ResponseBody
+    @ExceptionHandler(ActionValidationException.class)
+    public BasicSpcexResponse conflict(ActionValidationException ex) {
+        BasicSpcexResponse response = new BasicSpcexResponse();
+        EnumMessage message = ex.getErrors().stream().findFirst().orElseThrow();
+        response.setCode(message.getSubject().getId());
+        response.setMessage(errorResolver.resolve(message));
+        return response;
+    }
+
+    @ResponseStatus(value = HttpStatus.NOT_FOUND)
+    @ResponseBody
+    @ExceptionHandler(NotFound404Exception.class)
+    public BasicSpcexResponse conflict(NotFound404Exception ex) {
+        BasicSpcexResponse response = new BasicSpcexResponse();
+        EnumMessage message = ex.getError();
+        response.setCode(message.getSubject().getId());
+        response.setMessage(errorResolver.resolve(message));
+        return response;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/RequestStatusController.java

@@ -0,0 +1,48 @@
+package ru.spcex.clearing.backendapi.controller.system;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.ResponseBody;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+import ru.spcex.clearing.backendapi.controller.response.system.RequestInfoResponse;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
+import ru.spcex.clearing.platform.messaging.service.RequestInfo;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+
+@Controller
+@RequestMapping("/requests")
+public class RequestStatusController {
+
+    private final Imdg<RequestInfo> imdg;
+
+    @Autowired
+    public RequestStatusController(ImdgProvider imdgProvider) {
+        this.imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
+    }
+
+    @ApiOperation(value = "get request status.")
+    @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = RequestInfoResponse.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)})
+    @RequestMapping(value = "/{id}", method = RequestMethod.GET)
+    @ResponseBody
+    public RequestInfoResponse getById(@ApiParam(value = "Идентификатор объекта", required = true, example = "1234")
+                              @PathVariable("id") Long id) {
+        RequestInfo singleObjectByID;
+        if (id == null || (singleObjectByID = imdg.getSingleObjectByID(id)) == null) {
+            RequestInfoResponse response = new RequestInfoResponse();
+            response.setCode(-1L);
+            response.setMessage("cannot find request with id='" + id + "'.");
+            return response;
+        }
+        RequestInfoResponse response = new RequestInfoResponse();
+        response.setStatus(singleObjectByID.getStatus());
+        return response;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/ActionValidationException.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.backendapi.errors;
 
 import ru.spcex.platform.utils.enumeration.EnumMessage;
+import ru.spcex.platform.utils.enumeration.IEnumId;
 
 import java.util.Collection;
 import java.util.List;
@@ -16,6 +17,10 @@ public class ActionValidationException extends RuntimeException {
         this.errors = List.of(error);
     }
 
+    public ActionValidationException(IEnumId subject, Object... args) {
+        this.errors = List.of(new EnumMessage(subject, args));
+    }
+
     public Collection<EnumMessage> getErrors() {
         return errors;
     }

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/BackEndError.java

@@ -6,7 +6,10 @@ public enum BackEndError implements IEnumId {
     ValidationError(3000L),
     UnknownJsonProperty(3001L),
     FailedToReadHttpMessage(3002L),
-    KeycloakRepeatedRoles(3003L);
+    KeycloakRepeatedRoles(3003L),
+    DictionaryNotFound(3004L),
+    ResourceNotFound(404L)
+    ;
     private final Long id;
 
     BackEndError(Long id) {

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/NotFound404Exception.java

@@ -0,0 +1,16 @@
+package ru.spcex.clearing.backendapi.errors;
+
+import ru.spcex.platform.utils.enumeration.EnumMessage;
+
+public class NotFound404Exception extends RuntimeException {
+
+    private final EnumMessage error;
+
+    public NotFound404Exception(String comment) {
+        this.error = new EnumMessage(BackEndError.ResourceNotFound, comment);
+    }
+
+    public EnumMessage getError() {
+        return error;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/RequestInfoAccepter.java

@@ -0,0 +1,48 @@
+package ru.spcex.clearing.backendapi.service;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.stereotype.Service;
+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.service.QueueConsumer;
+import ru.spcex.clearing.platform.messaging.service.RequestInfo;
+import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+
+@Service
+public class RequestInfoAccepter extends QueueConsumer implements InitializingBean {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final Imdg<RequestInfo> requestInfoImdg;
+
+
+    public RequestInfoAccepter(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
+        super(kafkaQueue);
+        this.requestInfoImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
+    }
+
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        callback(RequestInfoUpdate.class)
+                .setConsumer(this::updateRequestInfo)
+                .forDestination(Consts.REQUEST_INFO_UPDATE, callbacks::put);
+        init();
+    }
+
+    private void updateRequestInfo(BaseRequest<RequestInfoUpdate> requestInfoUpdateBaseRequest) {
+        //will throw exception for any class other than RequestInfoUpdate
+        RequestInfoUpdate statusInfo = requestInfoUpdateBaseRequest.getRequestPayload();
+        RequestInfo requestInfo = requestInfoImdg.getSingleObjectByID(statusInfo.getId());
+        if (requestInfo == null) {
+            log.warn("unknown requestInfo id={}", statusInfo.getId());
+            return;
+        }
+        requestInfo.setStatus(statusInfo.getStatus());
+        requestInfoImdg.update(requestInfo);
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java

@@ -8,6 +8,7 @@ import ru.spcex.clearing.backendapi.controller.response.cud.QueueSuccessResponse
 import ru.spcex.clearing.backendapi.domain.actions.IAction;
 import ru.spcex.clearing.backendapi.errors.ActionValidationException;
 import ru.spcex.clearing.backendapi.service.IOperator;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
 import ru.spcex.platform.imdg.api.Imdg;
@@ -46,7 +47,7 @@ public class OperatorImpl implements IOperator {
     }
 
     private void saveRequestToStorage(String destination, BaseRequest<Object> request) {
-        Imdg<RequestInfo> requestStorage = imdgProvider.getImdg(destination, RequestInfo.class);
+        Imdg<RequestInfo> requestStorage = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
         RequestInfo requestInfo = RequestInfo.create(request.getId());
         requestStorage.insert(requestInfo);
     }

clearing-parent/backend-api/src/main/resources/application.properties

@@ -12,14 +12,22 @@ backend-api.hazelcast.cluster-members=127.0.0.1:5701
 backend-api.hazelcast.login=dev
 backend-api.hazelcast.password=dev-pass
 
-backend-api.kafka.bootstrap-servers=localhost:9092
+backend-api.kafka-producer.bootstrap-servers=localhost:9092
-backend-api.kafka.acks=all
+backend-api.kafka-producer.acks=all
-backend-api.kafka.retries=0
+backend-api.kafka-producer.retries=0
-backend-api.kafka.batch-size=16384
+backend-api.kafka-producer.batch-size=16384
-backend-api.kafka.linger-ms=1
+backend-api.kafka-producer.linger-ms=1
-backend-api.kafka.buffer-memory=33554432
+backend-api.kafka-producer.buffer-memory=33554432
 
-backend-api.security.authorization-disabled=true
+backend-api.kafka-consumer.bootstrap-servers=localhost:9092
+backend-api.kafka-consumer.group-id=dev-group-backend-api
+backend-api.kafka-consumer.enable-auto-commit=true
+backend-api.kafka-consumer.session-timeout-ms=30000
+backend-api.kafka-consumer.auto-offset-reset=latest
+backend-api.kafka-consumer.linger-ms=1
+backend-api.kafka-consumer.buffer-memory=33554432
+
+backend-api.security.authorization-disabled=false
 
 
 ##keycloak

clearing-parent/backend-api/src/main/resources/messages/error.properties

@@ -1,4 +1,5 @@
 1=This is error example 1.
 3000=validation error field '%s'
 3001=unrecognized json property '%s'
-3002=failed to read http message
+3002=failed to read http message
+3004=couldn't find dictionary '%s'

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

@@ -4,6 +4,7 @@ import org.apache.kafka.clients.producer.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+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.platform.imdg.api.Imdg;
@@ -22,7 +23,7 @@ public class KafkaSenderConfig {
                 .producer(kafkaProducer)
                 .idGenerator(imdgIdGenerator::nextId)
                 .imdgProvider(s -> {
-                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
+                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
                     return imdg::insert;
                 })
                 .build();

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

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.company.config;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.annotation.Bean;
@@ -8,6 +9,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
 import ru.spcex.clearing.company.config.settings.CompanyServiceSettings;
 import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 
 @Configuration
 public class KafkaConfig {
@@ -15,6 +17,13 @@ public class KafkaConfig {
     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
     @Bean
     public Consumer<String, Object> createConsumer(CompanyServiceSettings settings) {
-        return KafkaConsumerFactory.consumer(settings.getKafka());
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
     }
+
+    @Autowired
+    @Bean
+    public Producer<String, Object> createProducer(CompanyServiceSettings settings) {
+        return KafkaProducerFactory.producer(settings.getKafkaProducer());
+    }
+
 }

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/settings/CompanyServiceSettings.java

@@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.PropertySource;
 import org.springframework.stereotype.Component;
 import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
 @Component
@@ -11,7 +12,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 @ConfigurationProperties("company-service")
 public class CompanyServiceSettings {
     private HazelcastClientParams hazelcast;
-    private KafkaConsumerSettings kafka;
+    private KafkaConsumerSettings kafkaConsumer;
+    private KafkaProducerSettings kafkaProducer;
 
     public HazelcastClientParams getHazelcast() {
         return hazelcast;
@@ -21,11 +23,19 @@ public class CompanyServiceSettings {
         this.hazelcast = hazelcast;
     }
 
-    public KafkaConsumerSettings getKafka() {
+    public KafkaConsumerSettings getKafkaConsumer() {
-        return kafka;
+        return kafkaConsumer;
     }
 
-    public void setKafka(KafkaConsumerSettings kafka) {
+    public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
-        this.kafka = kafka;
+        this.kafkaConsumer = kafkaConsumer;
+    }
+
+    public KafkaProducerSettings getKafkaProducer() {
+        return kafkaProducer;
+    }
+
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
     }
 }

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ClearingMemberCategoryService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -20,9 +21,9 @@ public class ClearingMemberCategoryService extends QueueConsumer implements Init
     private final Imdg<ClearingMemberCategory> clearingMemberCategoryMap;
 
     @Autowired
-    public ClearingMemberCategoryService(Consumer<String, Object> kafkaQueue,
+    public ClearingMemberCategoryService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
                                          ImdgProvider imdgProvider) {
-        super(kafkaQueue);
+        super(kafkaQueue, kafkaProducer);
         this.clearingMemberCategoryMap = imdgProvider.getImdg(IMDGDistributedNames.Map_ClearingMemberCategory, ClearingMemberCategory.class);
     }
 

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyInfoService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -23,9 +24,9 @@ public class CompanyInfoService extends QueueConsumer implements InitializingBea
     private final Imdg<Company> companyMap;
 
     @Autowired
-    public CompanyInfoService(Consumer<String, Object> kafkaQueue,
+    public CompanyInfoService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
                               ImdgProvider imdgProvider) {
-        super(kafkaQueue);
+        super(kafkaQueue, kafkaProducer);
 //        this.companyInfoMap = imdgProvider.getImdg(IMDGDistributedNames.Map_CompanyInfo, CompanyInfo.class);
         this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class);
     }

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -21,9 +22,9 @@ public class CompanyService extends QueueConsumer implements InitializingBean {
     private final Imdg<Company> companyMap;
 
     @Autowired
-    public CompanyService(Consumer<String, Object> kafkaQueue,
+    public CompanyService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
                           ImdgProvider imdgProvider) {
-        super(kafkaQueue);
+        super(kafkaQueue, kafkaProducer);
         this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class);
     }
 

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanySymbolService.java

@@ -2,6 +2,7 @@ package ru.spcex.clearing.company.service;
 
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -22,9 +23,9 @@ public class CompanySymbolService extends QueueConsumer implements InitializingB
     private final Imdg<CompanySymbols> companySymbolsMap;
 
     @Autowired
-    public CompanySymbolService(Consumer<String, Object> kafkaQueue,
+    public CompanySymbolService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
                                 ImdgProvider imdgProvider) {
-        super(kafkaQueue);
+        super(kafkaQueue, kafkaProducer);
         this.companySymbolsMap = imdgProvider.getImdg(IMDGDistributedNames.Map_CompanySymbols, CompanySymbols.class);
     }
 

clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ContactService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.company.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -19,9 +20,9 @@ public class ContactService extends QueueConsumer implements InitializingBean {
     private final Imdg<Contact> contactMap;
 
     @Autowired
-    public ContactService(Consumer<String, Object> kafkaQueue,
+    public ContactService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
                           ImdgProvider imdgProvider) {
-        super(kafkaQueue);
+        super(kafkaQueue, kafkaProducer);
         this.contactMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Contact, Contact.class);
     }
 

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

@@ -2,10 +2,17 @@ spring.main.web-application-type=none
 company-service.hazelcast.cluster-members=127.0.0.1:5701
 company-service.hazelcast.login=dev
 company-service.hazelcast.password=dev-pass
-company-service.kafka.bootstrap-servers=localhost:9092
+company-service.kafka-consumer.bootstrap-servers=localhost:9092
-company-service.kafka.group-id=dev-group-company-service
+company-service.kafka-consumer.group-id=dev-group-company-service
-company-service.kafka.enable-auto-commit=true
+company-service.kafka-consumer.enable-auto-commit=true
-company-service.kafka.session-timeout-ms=30000
+company-service.kafka-consumer.session-timeout-ms=30000
-company-service.kafka.auto-offset-reset=latest
+company-service.kafka-consumer.auto-offset-reset=latest
-company-service.kafka.linger-ms=1
+company-service.kafka-consumer.linger-ms=1
-company-service.kafka.buffer-memory=33554432
+company-service.kafka-consumer.buffer-memory=33554432
+
+company-service.kafka-producer.bootstrap-servers=localhost:9092
+company-service.kafka-producer.acks=all
+company-service.kafka-producer.retries=0
+company-service.kafka-producer.batch-size=16384
+company-service.kafka-producer.linger-ms=1
+company-service.kafka-producer.buffer-memory=33554432

clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java

@@ -5,6 +5,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
@@ -28,7 +29,7 @@ public class KafkaConfig {
                     .producer(kafkaProducer)
                     .idGenerator(imdgIdGenerator::nextId)
                     .imdgProvider(s -> {
-                        Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
+                        Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
                         return imdg::insert;
                     })
                     .build();

clearing-parent/dictionary/pom.xml

@@ -55,6 +55,10 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-classes-base</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

clearing-parent/dictionary/src/main/java/ru/clearing/platform/dictionary/AbstractDictionary.java

@@ -1,22 +1,12 @@
 package ru.clearing.platform.dictionary;
 
-public abstract class AbstractDictionary implements Dictionary {
+import ru.spcex.platform.classes.base.SpcexObjectBase;
-    static final long serialVersionUID = ConstDictionarySerializable.serialVersionUID;
 
-    protected Long id;
+public abstract class AbstractDictionary extends SpcexObjectBase implements Dictionary {
+    static final long serialVersionUID = ConstDictionarySerializable.serialVersionUID;
     protected String code;
     protected String name;
 
-    @Override
-    public Long getId() {
-        return id;
-    }
-
-    @Override
-    public void setId(Long id) {
-        this.id = id;
-    }
-
     @Override
     public String getCode() {
         return code;

clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/dictionary/ClearingMemberCategoryDictionaryMapStore.java

@@ -0,0 +1,31 @@
+package ru.spcex.clearing.imdg.dictionary;
+
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+import ru.clearing.platform.dictionary.ClearingMemberCategoryDictionary;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
+import ru.spcex.clearing.imdg.base.DictionaryTMapStore;
+
+@Component
+public class ClearingMemberCategoryDictionaryMapStore extends DictionaryTMapStore<ClearingMemberCategoryDictionary> {
+
+    public ClearingMemberCategoryDictionaryMapStore(JdbcTemplate jdbcTemplate) {
+        super(jdbcTemplate);
+    }
+
+    @Override
+    public String getMapName() {
+        return IMDGDistributedNames.Map_ClearingMemberCategoryDictionary;
+    }
+
+    @Override
+    public String getTableName() {
+        return "CLEARING_MEMBER_CATEGORY_DICTIONARY";
+    }
+
+    @Override
+    public ClearingMemberCategoryDictionary getDictionaryObject() {
+        return new ClearingMemberCategoryDictionary();
+    }
+
+}

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

@@ -1,10 +1,12 @@
 package ru.spcex.clearing.securities.config;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 import ru.spcex.clearing.securities.config.element.SecuritiesServiceSettings;
 
 @Configuration
@@ -12,8 +14,15 @@ public class KafkaConfig {
 
     @Autowired
     @Bean
-    public Consumer<String, Object> createProducer(SecuritiesServiceSettings settings) {
+    public Consumer<String, Object> createConsumer(SecuritiesServiceSettings settings) {
-        return KafkaConsumerFactory.consumer(settings.getKafka());
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
     }
 
+    @Autowired
+    @Bean
+    public Producer<String, Object> createProducer(SecuritiesServiceSettings settings) {
+        return KafkaProducerFactory.producer(settings.getKafkaProducer());
+    }
+
+
 }

clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/element/SecuritiesServiceSettings.java

@@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.PropertySource;
 import org.springframework.stereotype.Component;
 import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
 @Component
@@ -11,7 +12,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 @ConfigurationProperties("securities-service")
 public class SecuritiesServiceSettings {
     private HazelcastClientParams hazelcast;
-    private KafkaConsumerSettings kafka;
+    private KafkaConsumerSettings kafkaConsumer;
+    private KafkaProducerSettings kafkaProducer;
 
     public HazelcastClientParams getHazelcast() {
         return hazelcast;
@@ -21,11 +23,19 @@ public class SecuritiesServiceSettings {
         this.hazelcast = hazelcast;
     }
 
-    public KafkaConsumerSettings getKafka() {
+    public KafkaConsumerSettings getKafkaConsumer() {
-        return kafka;
+        return kafkaConsumer;
     }
 
-    public void setKafka(KafkaConsumerSettings kafka) {
+    public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
-        this.kafka = kafka;
+        this.kafkaConsumer = kafkaConsumer;
+    }
+
+    public KafkaProducerSettings getKafkaProducer() {
+        return kafkaProducer;
+    }
+
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
     }
 }

clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.securities.service.cud;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -19,7 +20,6 @@ import ru.spcex.clearing.platform.messaging.service.Status;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgProvider;
 import ru.spcex.platform.imdg.api.ImdgTransaction;
-import ru.spcex.platform.utils.log.ExceptionUtils;
 
 import java.math.BigDecimal;
 
@@ -29,15 +29,16 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
     private final Imdg<MoneyMarketSecurity> moneyMarketSecurityMap;
     private final ImdgProvider imdgProvider;
     @Autowired
-    public MoneyMarketSecurityService(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
+    public MoneyMarketSecurityService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
-        super(kafkaQueue);
+                                      ImdgProvider imdgProvider) {
+        super(kafkaQueue, kafkaProducer);
         this.moneyMarketSecurityMap = imdgProvider.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class);
         this.imdgProvider = imdgProvider;
     }
 
     @Override
     protected boolean needsProcessing(String destination, BaseRequest<?> request) {
-        Imdg<RequestInfo> requestSpecificImdg = imdgProvider.getImdg(destination, RequestInfo.class);
+        Imdg<RequestInfo> requestSpecificImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
         RequestInfo requestFound = requestSpecificImdg.getSingleObjectByID(request.getId());
         if (requestFound == null) {
             log.error("couldn't find request info by id={}, action type={}", request.getId(), request.getActionType());
@@ -66,8 +67,8 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
     }
 
     private void newMoneyMarket(BaseRequest<MoneyMarketSecurityNewRequest> userRequest) {
-        Imdg<RequestInfo> requestInfoImdg = imdgProvider.getImdg(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, RequestInfo.class);
+//        Imdg<RequestInfo> requestInfoImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
-        RequestInfo reqInfo = requestInfoImdg.getSingleObjectByID(userRequest.getId());
+//        RequestInfo reqInfo = requestInfoImdg.getSingleObjectByID(userRequest.getId());
         ImdgTransaction transaction = imdgProvider.newTransaction();
         MoneyMarketSecurityNewRequest req = userRequest.getRequestPayload();
         log.debug("MoneyMarketSecurityNewRequest received");
@@ -82,17 +83,17 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
         try {
             transaction.beginTransaction();
             Imdg<MoneyMarketSecurity> moneyMarketSecurityMap = transaction.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class);
-            Imdg<RequestInfo> reqInfoMap = transaction.getImdg(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, RequestInfo.class);
+            Imdg<RequestInfo> reqInfoMap = transaction.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
             //одним скопом выполняем реквест
             moneyMarketSecurityMap.insert(mms);
-            RequestInfo.update(reqInfo, Status.Success, "success");
+//            RequestInfo.update(reqInfo, Status.Success, "success");
             //и сохраняем обновленный
-            reqInfoMap.insert(reqInfo);
+//            reqInfoMap.insert(reqInfo);
             transaction.commitTransaction();
         } catch (Throwable e) {
             transaction.rollbackTransaction();
-            RequestInfo.update(reqInfo, Status.Error, ExceptionUtils.getStackTrace(e));
+//            RequestInfo.update(reqInfo, Status.Error, ExceptionUtils.getStackTrace(e));
-            requestInfoImdg.insert(reqInfo);
+//            requestInfoImdg.insert(reqInfo);
         }
         log.debug("successfully processed, request id {}, new object id {}", userRequest.getId(), mms.getId());
     }

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

@@ -4,13 +4,20 @@ securities-service.hazelcast.cluster-members=127.0.0.1:5701
 securities-service.hazelcast.login=dev
 securities-service.hazelcast.password=dev-pass
 
-securities-service.kafka.bootstrap-servers=localhost:9092
+securities-service.kafka-consumer.bootstrap-servers=localhost:9092
-securities-service.kafka.group-id=dev-group
+securities-service.kafka-consumer.group-id=dev-group-securities-service
-securities-service.kafka.enable-auto-commit=false
+securities-service.kafka-consumer.enable-auto-commit=false
-securities-service.kafka.session-timeout-ms=30000
+securities-service.kafka-consumer.session-timeout-ms=30000
-securities-service.kafka.auto-offset-reset=latest
+securities-service.kafka-consumer.auto-offset-reset=latest
-securities-service.kafka.linger-ms=1
+securities-service.kafka-consumer.linger-ms=1
-securities-service.kafka.buffer-memory=33554432
+securities-service.kafka-consumer.buffer-memory=33554432
+
+securities-service.kafka-producer.bootstrap-servers=localhost:9092
+securities-service.kafka-producer.acks=all
+securities-service.kafka-producer.retries=0
+securities-service.kafka-producer.batch-size=16384
+securities-service.kafka-producer.linger-ms=1
+securities-service.kafka-producer.buffer-memory=33554432
 #props.put("bootstrap.servers", kafkaSettings.getBootstrapServers());
 #if (kafkaSettings.getGroupId() != null && kafkaSettings.getGroupId().length() > 0) {
 #    props.put("group.id", kafkaSettings.getGroupId());

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

@@ -1,12 +1,14 @@
 package ru.spcex.clearing.utility.config;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
 import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 import ru.spcex.clearing.utility.config.settings.UtilityServiceSettings;
 
 @Configuration
@@ -14,7 +16,14 @@ public class KafkaConfig {
     @Autowired
     @Bean
     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
-    public Consumer<String, Object> createProducer(UtilityServiceSettings settings) {
+    public Consumer<String, Object> createConsumer(UtilityServiceSettings settings) {
-        return KafkaConsumerFactory.consumer(settings.getKafka());
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
     }
+
+    @Autowired
+    @Bean
+    public Producer<String, Object> createProducer(UtilityServiceSettings settings) {
+        return KafkaProducerFactory.producer(settings.getKafkaProducer());
+    }
+
 }

clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/settings/UtilityServiceSettings.java

@@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.PropertySource;
 import org.springframework.stereotype.Component;
 import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
 @Component
@@ -11,7 +12,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 @ConfigurationProperties("utility-service")
 public class UtilityServiceSettings {
     private HazelcastClientParams hazelcast;
-    private KafkaConsumerSettings kafka;
+    private KafkaConsumerSettings kafkaConsumer;
+    private KafkaProducerSettings kafkaProducer;
 
     public HazelcastClientParams getHazelcast() {
         return hazelcast;
@@ -21,11 +23,19 @@ public class UtilityServiceSettings {
         this.hazelcast = hazelcast;
     }
 
-    public KafkaConsumerSettings getKafka() {
+    public KafkaConsumerSettings getKafkaConsumer() {
-        return kafka;
+        return kafkaConsumer;
     }
 
-    public void setKafka(KafkaConsumerSettings kafka) {
+    public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
-        this.kafka = kafka;
+        this.kafkaConsumer = kafkaConsumer;
+    }
+
+    public KafkaProducerSettings getKafkaProducer() {
+        return kafkaProducer;
+    }
+
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
     }
 }

clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/KeyRateService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.utility.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -23,8 +24,9 @@ public class KeyRateService extends QueueConsumer implements InitializingBean {
     private final Imdg<KeyRate> keyRateMap;
 
     @Autowired
-    public KeyRateService(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
+    public KeyRateService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
-        super(kafkaQueue);
+                          ImdgProvider imdgProvider) {
+        super(kafkaQueue, kafkaProducer);
         this.keyRateMap = imdgProvider.getImdg(IMDGDistributedNames.Map_KeyRate, KeyRate.class);
     }
 

clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/UserSettingService.java

@@ -1,6 +1,7 @@
 package ru.spcex.clearing.utility.service;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -23,8 +24,9 @@ public class UserSettingService extends QueueConsumer implements InitializingBea
     private final Imdg<UserSettings> userSettingsMap;
 
     @Autowired
-    public UserSettingService(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
+    public UserSettingService(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
-        super(kafkaQueue);
+                              ImdgProvider imdgProvider) {
+        super(kafkaQueue, kafkaProducer);
         this.userSettingsMap = imdgProvider.getImdg(IMDGDistributedNames.Map_UserSettings, UserSettings.class);
     }
 

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

@@ -4,10 +4,17 @@ utility-service.hazelcast.cluster-members=127.0.0.1:5701
 utility-service.hazelcast.login=dev
 utility-service.hazelcast.password=dev-pass
 
-utility-service.kafka.bootstrap-servers=localhost:9092
+utility-service.kafka-consumer.bootstrap-servers=localhost:9092
-utility-service.kafka.group-id=dev-group-utility-service
+utility-service.kafka-consumer.group-id=dev-group-utility-service
-utility-service.kafka.enable-auto-commit=true
+utility-service.kafka-consumer.enable-auto-commit=true
-utility-service.kafka.session-timeout-ms=30000
+utility-service.kafka-consumer.session-timeout-ms=30000
-utility-service.kafka.auto-offset-reset=latest
+utility-service.kafka-consumer.auto-offset-reset=latest
-utility-service.kafka.linger-ms=1
+utility-service.kafka-consumer.linger-ms=1
-utility-service.kafka.buffer-memory=33554432
+utility-service.kafka-consumer.buffer-memory=33554432
+
+utility-service.kafka-producer.bootstrap-servers=localhost:9092
+utility-service.kafka-producer.acks=all
+utility-service.kafka-producer.retries=0
+utility-service.kafka-producer.batch-size=16384
+utility-service.kafka-producer.linger-ms=1
+utility-service.kafka-producer.buffer-memory=33554432

platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java

@@ -29,6 +29,11 @@ public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
         return map.values();
     }
 
+    @Override
+    public Integer size() {
+        return map.size();
+    }
+
     @Override
     public void insert(T paramT) {
         if (paramT.getId() == null) {

platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java

@@ -101,6 +101,7 @@ public final class IMDGDistributedNames {
     public static final String Map_AccountStatusDictionary = "Map_AccountStatusDictionary";
     public static final String Map_KeyRate = "Map_KeyRate";
     public static final String Map_MoneyMarketSecurity = "Map_MoneyMarketSecurity";
+    public static final String Map_RequestInfo = "Map_RequestInfo";
     public static final String MAP_SEQUENCE_NAME = "MAP_SEQUENCE_NAME"; // todo вынести idGenerator отдельно и завернуть в метод, чтобы не напрямую обращаться.
 
     private IMDGDistributedNames() {

platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/Imdg.java

@@ -57,4 +57,8 @@ public interface Imdg<T extends SpcexObjectBase> {
     default <A> Collection<A> projectionsAttributeBySql(String paramString, String... paramVarArgs) {
         throw new UnsupportedOperationException("not implemented projectionsAttributeBySql");
     }
+
+    default Integer size() {
+        throw new UnsupportedOperationException("not implemented size");
+    }
 }

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

@@ -33,4 +33,6 @@ public interface Consts {
     String ACCOUNT_NEW = "account-new";
     String BALANCE_ACCOUNT_NEW = "balance-account-new";
 
+    String REQUEST_INFO_UPDATE = "request-info-update";
+
 }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/serialize/EnumSerializer.java

@@ -0,0 +1,19 @@
+package ru.spcex.clearing.platform.messaging.domain.json.serialize;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.time.format.DateTimeFormatter;
+
+public class EnumSerializer extends JsonSerializer<Enum<?>> {
+    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+
+
+    @Override
+    public void serialize(Enum<?> value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+        if (value == null) return;
+        gen.writeString(value.name());
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java

@@ -3,7 +3,9 @@ package ru.spcex.clearing.platform.messaging.logic.functional;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 public interface BuilderConsumerStep<T1> {
     BuilderDestinationStep setConsumer(Consumer<BaseRequest<T1>> consumer);
+    BuilderDestinationStep setFunction(Function<BaseRequest<T1>, Object> function);
 }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java

@@ -4,6 +4,7 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 /**
  * вспомогательный класс
@@ -18,6 +19,7 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
      * действие-обработчик реквеста (метод из конкретного менеджера)
      */
     private java.util.function.Consumer<BaseRequest<T>> consumer;
+    private java.util.function.Function<BaseRequest<T>, Object> function;
 
     private ConsumerSpecificClass(Class<T> clazz) {
         this.clazz = clazz;
@@ -29,7 +31,17 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
 
     @Override
     public BuilderDestinationStep setConsumer(Consumer<BaseRequest<T>> consumer) {
-        this.consumer = consumer;
+        if (this.consumer == null) {
+            this.consumer = consumer;
+        } else {
+            this.consumer = this.consumer.andThen(consumer);
+        }
+        return this;
+    }
+
+    @Override
+    public BuilderDestinationStep setFunction(Function<BaseRequest<T>, Object> function) {
+        this.function = function;
         return this;
     }
 
@@ -38,10 +50,18 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
     }
 
     @SuppressWarnings("unchecked")
-    public void acceptRaw(Object obj) {
+    public Object acceptRaw(Object obj) {
-        this.consumer.accept((BaseRequest<T>) obj);
+        if (this.consumer != null) {
+            this.consumer.accept((BaseRequest<T>) obj);
+            return null;
+        } else {
+            return this.function.apply((BaseRequest<T>) obj);
+        }
     }
 
+    public Consumer<BaseRequest<T>> getConsumer() {
+        return consumer;
+    }
     public Class<T> getClazz() {
         return clazz;
     }

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

@@ -5,10 +5,15 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+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.errors.WakeupException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import ru.spcex.clearing.platform.messaging.domain.ActionType;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.logic.functional.BuilderConsumerStep;
 import ru.spcex.clearing.platform.messaging.logic.functional.ConsumerSpecificClass;
 import ru.spcex.platform.utils.log.ExceptionUtils;
@@ -19,6 +24,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -28,33 +34,40 @@ public class QueueConsumer implements AutoCloseable {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final Consumer<String, Object> consumer;
-    private final ExecutorService executor;
+    private Producer<String, Object> producer;
+    private final ExecutorService inputExecutor;
+    private final ExecutorService outputExecutor;
     protected final Map<String, ConsumerSpecificClass<?>> callbacks;
     private final ObjectMapper json;
+    protected boolean supportStartOffsetTimeWindow;
 
     public QueueConsumer(Consumer<String, Object> kafkaQueue) {
         this.consumer = kafkaQueue;
         this.callbacks = new HashMap<>();
-        this.executor = Executors.newSingleThreadExecutor();
+        this.inputExecutor = Executors.newSingleThreadExecutor();
+        this.outputExecutor = Executors.newSingleThreadExecutor();
         this.json = new ObjectMapper();
+        this.supportStartOffsetTimeWindow = false;
+    }
+
+    public QueueConsumer(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaResponseQueue) {
+        this(kafkaQueue);
+        this.producer = kafkaResponseQueue;
     }
 
     protected boolean needsProcessing(String topicName, BaseRequest<?> request) {
         return true;
     }
 
-    protected boolean supportStartOffsetTimeWindow() {
-        return false;
-    }
-
     public void init() {
-        executor.submit(() -> {
+        inputExecutor.submit(() -> {
             try {
-                if (supportStartOffsetTimeWindow()) {
+                if (supportStartOffsetTimeWindow) {
                     consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet()));
                 } else {
                     consumer.subscribe(callbacks.keySet());
                 }
+                Object o = null;
                 while (!closed.get()) {
                     try {
                         ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
@@ -62,13 +75,19 @@ public class QueueConsumer implements AutoCloseable {
                             ConsumerSpecificClass<?> callback = callbacks.get(next.topic());
                             Class<?> clazz = callback.getClazz();
                             JavaType payloadType = json.getTypeFactory().constructParametricType(BaseRequest.class, clazz);
-                            Object o = json.readValue((String) next.value(), payloadType);
+                            o = json.readValue((String) next.value(), payloadType);
                             if (needsProcessing(next.topic(), (BaseRequest<?>) o)) {
-                                callback.acceptRaw(o);
+                                Object topicResponse = callback.acceptRaw(o);
+                                if (producer != null) {
+                                    sendResponse((BaseRequest<?>) o, topicResponse);
+                                }
                             }
                         }
                     } catch (Throwable e) {
                         log.error(ExceptionUtils.getStackTrace(e));
+                        if (producer != null) {
+                            sendErrorResponse((BaseRequest<?>) o);
+                        }
                     }
                 }
             } catch (WakeupException e) {
@@ -81,6 +100,51 @@ public class QueueConsumer implements AutoCloseable {
         });
     }
 
+    //мб перенести в другой класс
+    private void sendErrorResponse(BaseRequest<?> o) {
+        outputExecutor.submit(() -> {
+            try {
+                Future<RecordMetadata> send;
+                //default response
+                BaseRequest<RequestInfoUpdate> req = new BaseRequest<>();
+                RequestInfoUpdate success = new RequestInfoUpdate();
+                success.setId(o.getId());
+                success.setStatus(Status.Error);
+                req.setRequestPayload(success);
+                req.setId(o.getId());
+                req.setActionType(ActionType.SYSTEM);
+                send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req));
+                send.get();
+            } catch (Exception e) {
+                log.error(ExceptionUtils.getStackTrace(e));
+            }
+        });
+    }
+
+    private void sendResponse(BaseRequest<?> o, Object response) {
+        outputExecutor.submit(() -> {
+            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);
+                }
+                send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req));
+                send.get();
+            } catch (Exception e) {
+                log.error(ExceptionUtils.getStackTrace(e));
+            }
+        });
+    }
+
     protected <T> BuilderConsumerStep<T> callback(Class<T> clazz) {
         return ConsumerSpecificClass.build(clazz);
     }

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

@@ -0,0 +1,22 @@
+package ru.spcex.clearing.platform.messaging.service;
+
+public class RequestInfoUpdate {
+    private Long id;
+    private Status status;
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    public Status getStatus() {
+        return status;
+    }
+
+    public void setStatus(Status status) {
+        this.status = status;
+    }
+}