IN Migration Kafka 011 Additional

012 013

Временное хранение смигрированных данных

014 015

Каждая мигрированная запись храниться в монго коллекций

016 017

018  mongo.mybpm_aux.KafkaMigrationRecord
019  
020  
021  
022

023 024

Это предотвращает перезапись данных.

025 026

027

Класс, который используется в этой, коллекций:

028 029

030  public class KafkaMigrationRecordDto
031  
032  
033  
034

035 036

Расположения данного класса:

037 038

039  kz/greetgo/mybpm/model_kafka_mongo/migration/KafkaMigrationRecordDto.java
040  
041  
042  
043

044 045

046

Schedulers of 047 InMigrationKafka

048 049

Все шедулеры связанные с миграцией через кафку находиться в 050 классе:

051 052

053  public class InMigrationKafkaScheduler
054  
055  
056

057 058

059

Удаление временных записей из монго коллекций

060 061

062     @Scheduled("#")
063      @FromConfig("repeat every 5 hour")
064      public void deleteRecords() {
065      inMigrationKafka.deleteKafkaMigrationRecord();
066      }
067  
068  
069

070 071

Повторяются каждые 5 часов, и удаляет просроченные 072 KafkaMigrationRecordDto. Время 073 простроченных записей можно определить в конфиге

074 075

076

Запуск дополнительной миграций для не найденных значение 077 ссылочных БО и СО

078 079

080     @Scheduled("#")
081      @FromConfig("repeat every 10 min")
082      public void applyRefBoi() {
083      inMigrationKafkaRefBoi.migrateFromTempTable();
084      }
085  
086  
087  
088

089 090

Запускается каждые 10 минут если находит значение то, смигрирует 091 эту запись и сделает запись не актуальным во временной таблице. 092 Примечание эти временные данные хранятся в ПОСТГРЕСЕ

093 094

095

Удаляет не актуальные записи в 23:30, под не актуальными 096 записями имеется ввиду уже смигрированные данные которые описан 097 выше

098 099

100  @Scheduled("#")
101  @FromConfig("23:30")
102  public void deleteUnActualRecords() {
103  inMigrationKafkaRefBoi.deleteUnActualRecords();
104  }
105  
106  
107  
108

109 110

111

Config of InMigrationKafka

112 113

114  @Description("Конфиг используемые во время миграции в через кафку")
115  public interface InMigrationKafkaConfig {
116  
117  @Description("Количество часов до очищения временных записей")
118  @DefaultIntValue(5)
119  int recordCleanHours();
120  
121  @Description("Интервал в минутах для обновление не просаженных ссылочных записей")
122  @DefaultIntValue(10)
123  long interval();
124  
125  @Description("Количество минут до удаления старых записей из Базы")
126  @DefaultIntValue(10080)
127  long oldRecordDeleteMinutes();
128  }
129  
130  
131  
132

133