LWDEV-9478 Refactor balances sub-system

src/dist/cfg/log4j2.xml

@@ -344,6 +344,24 @@
             <AppenderRef ref="ConsoleLog"/>
         </Logger>
 
+        <Logger name="com.lykke.matching.engine.database.reconciliation.AccountsMigrationService"
+                     level="info" additivity="false">
+            <AppenderRef ref="ServerLog"/>
+            <AppenderRef ref="ConsoleLog"/>
+        </Logger>
+
+        <Logger name="com.lykke.matching.engine.database.reconciliation.OrdersMigrationService"
+                     level="info" additivity="false">
+            <AppenderRef ref="ServerLog"/>
+            <AppenderRef ref="ConsoleLog"/>
+        </Logger>
+
+        <Logger name="com.lykke.matching.engine.utils.balance.ReservedVolumesRecalculator"
+                     level="info" additivity="false">
+            <AppenderRef ref="ServerLog"/>
+            <AppenderRef ref="ConsoleLog"/>
+        </Logger>
+
         <AsyncLogger name="com.lykke.matching.engine.services.events.listeners.MessageProcessingSwitchListener"
                      level="info" additivity="false">
             <AppenderRef ref="ServerLog"/>
@@ -541,6 +559,16 @@
             <AppenderRef ref="BalanceUpdatesLog"/>
         </AsyncLogger>
 
+        <AsyncLogger name="com.lykke.matching.engine.services.BalancesServiceImpl"
+                     level="info" additivity="false">
+            <AppenderRef ref="BalanceUpdatesLog"/>
+        </AsyncLogger>
+
+        <AsyncLogger name="com.lykke.matching.engine.outgoing.senders.impl.specialized.CashInOutOldEventSender"
+                     level="info" additivity="false">
+            <AppenderRef ref="BalanceUpdatesLog"/>
+        </AsyncLogger>
+
         # Errors logger
         <AsyncLogger name="com.lykke.utils.logging.MetricsLogger"
                      level="debug" additivity="false">

src/main/kotlin/com/lykke/matching/engine/balance/WalletOperationsProcessor.kt

@@ -3,15 +3,9 @@ package com.lykke.matching.engine.balance
 import com.lykke.matching.engine.daos.WalletOperation
 import com.lykke.matching.engine.daos.wallet.AssetBalance
 import com.lykke.matching.engine.daos.wallet.Wallet
-import com.lykke.matching.engine.database.PersistenceManager
 import com.lykke.matching.engine.database.common.entity.BalancesData
-import com.lykke.matching.engine.database.common.entity.OrderBooksPersistenceData
-import com.lykke.matching.engine.database.common.entity.PersistenceData
-import com.lykke.matching.engine.deduplication.ProcessedMessage
 import com.lykke.matching.engine.holders.ApplicationSettingsHolder
 import com.lykke.matching.engine.holders.AssetsHolder
-import com.lykke.matching.engine.holders.BalancesHolder
-import com.lykke.matching.engine.outgoing.messages.BalanceUpdate
 import com.lykke.matching.engine.outgoing.messages.ClientBalanceUpdate
 import com.lykke.matching.engine.order.transaction.CurrentTransactionBalancesHolder
 import com.lykke.matching.engine.order.transaction.WalletAssetBalance
@@ -20,21 +14,18 @@ import com.lykke.utils.logging.MetricsLogger
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import java.math.BigDecimal
-import java.util.Date
 
-class WalletOperationsProcessor(private val balancesHolder: BalancesHolder,
-                                private val currentTransactionBalancesHolder: CurrentTransactionBalancesHolder,
+class WalletOperationsProcessor(private val currentTransactionBalancesHolder: CurrentTransactionBalancesHolder,
                                 private val applicationSettingsHolder: ApplicationSettingsHolder,
-                                private val persistenceManager: PersistenceManager,
                                 private val assetsHolder: AssetsHolder,
-                                private val logger: Logger?): BalancesGetter {
+                                private val logger: Logger?) : BalancesGetter {
 
     companion object {
         private val LOGGER = LoggerFactory.getLogger(WalletOperationsProcessor::class.java.name)
         private val METRICS_LOGGER = MetricsLogger.getLogger()
     }
 
-   private val clientBalanceUpdatesByClientIdAndAssetId = HashMap<String, ClientBalanceUpdate>()
+    private val clientBalanceUpdatesByClientIdAndAssetId = HashMap<String, ClientBalanceUpdate>()
 
     fun preProcess(operations: Collection<WalletOperation>,
                    allowInvalidBalance: Boolean = false,
@@ -60,6 +51,14 @@ class WalletOperationsProcessor(private val balancesHolder: BalancesHolder,
                 changedAssetBalance.reserved
         }
 
+        validateChangedAssetBalances(changedAssetBalances, allowInvalidBalance)
+
+
+        changedAssetBalances.forEach { processChangedAssetBalance(it.value) }
+        return this
+    }
+
+    private fun validateChangedAssetBalances(changedAssetBalances: HashMap<String, ChangedAssetBalance>, allowInvalidBalance: Boolean) {
         try {
             changedAssetBalances.values.forEach { validateBalanceChange(it) }
         } catch (e: BalanceException) {
@@ -70,9 +69,6 @@ class WalletOperationsProcessor(private val balancesHolder: BalancesHolder,
             (logger ?: LOGGER).error(message)
             METRICS_LOGGER.logError(message, e)
         }
-
-        changedAssetBalances.forEach { processChangedAssetBalance(it.value) }
-        return this
     }
 
     private fun processChangedAssetBalance(changedAssetBalance: ChangedAssetBalance) {
@@ -114,23 +110,6 @@ class WalletOperationsProcessor(private val balancesHolder: BalancesHolder,
         return currentTransactionBalancesHolder.persistenceData()
     }
 
-    fun persistBalances(processedMessage: ProcessedMessage?,
-                        orderBooksData: OrderBooksPersistenceData?,
-                        stopOrderBooksData: OrderBooksPersistenceData?,
-                        messageSequenceNumber: Long?): Boolean {
-        return persistenceManager.persist(PersistenceData(persistenceData(),
-                processedMessage,
-                orderBooksData,
-                stopOrderBooksData,
-                messageSequenceNumber))
-    }
-
-    fun sendNotification(id: String, type: String, messageId: String) {
-        if (clientBalanceUpdatesByClientIdAndAssetId.isNotEmpty()) {
-            balancesHolder.sendBalanceUpdate(BalanceUpdate(id, type, Date(), clientBalanceUpdatesByClientIdAndAssetId.values.toList(), messageId))
-        }
-    }
-
     fun getClientBalanceUpdates(): List<ClientBalanceUpdate> {
         return clientBalanceUpdatesByClientIdAndAssetId.values.toList()
     }

src/main/kotlin/com/lykke/matching/engine/balance/WalletOperationsProcessorFactory.kt

@@ -0,0 +1,21 @@
+package com.lykke.matching.engine.balance
+
+import com.lykke.matching.engine.holders.ApplicationSettingsHolder
+import com.lykke.matching.engine.holders.AssetsHolder
+import com.lykke.matching.engine.order.transaction.CurrentTransactionBalancesHolderFactory
+import org.slf4j.Logger
+import org.springframework.stereotype.Component
+
+@Component
+class WalletOperationsProcessorFactory(private val currentTransactionBalancesHolderFactory: CurrentTransactionBalancesHolderFactory,
+                                       private val applicationSettingsHolder: ApplicationSettingsHolder,
+                                       private val assetsHolder: AssetsHolder) {
+    fun create(logger: Logger?): WalletOperationsProcessor {
+
+        return WalletOperationsProcessor(
+                currentTransactionBalancesHolderFactory.create(),
+                applicationSettingsHolder,
+                assetsHolder,
+                logger)
+    }
+}

src/main/kotlin/com/lykke/matching/engine/database/reconciliation/AccountsMigrationService.kt

@@ -5,7 +5,7 @@ import com.lykke.matching.engine.database.Storage
 import com.lykke.matching.engine.database.azure.AzureWalletDatabaseAccessor
 import com.lykke.matching.engine.database.redis.accessor.impl.RedisWalletDatabaseAccessor
 import com.lykke.matching.engine.exception.MatchingEngineException
-import com.lykke.matching.engine.holders.BalancesHolder
+import com.lykke.matching.engine.services.BalancesService
 import com.lykke.matching.engine.utils.config.Config
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -17,9 +17,9 @@ import java.util.*
 
 @Component
 @Order(1)
-class AccountsMigrationService @Autowired constructor (private val balancesHolder: BalancesHolder,
-                                                       private val config: Config,
-                                                       private val redisWalletDatabaseAccessor: Optional<RedisWalletDatabaseAccessor>): ApplicationRunner {
+class AccountsMigrationService @Autowired constructor (private val config: Config,
+                                                       private val redisWalletDatabaseAccessor: Optional<RedisWalletDatabaseAccessor>,
+                                                       private val balancesService: BalancesService): ApplicationRunner {
     override fun run(args: ApplicationArguments?) {
         if (config.me.walletsMigration) {
             migrateAccounts()
@@ -55,38 +55,41 @@ class AccountsMigrationService @Autowired constructor (private val balancesHolde
         }
 
         val startTime = Date().time
-        teeLog("Starting wallets migration from azure to redis; azure table: $azureAccountsTableName, redis: ${config.me.redis.host}.${config.me.redis.port}")
+        LOGGER.info("Starting wallets migration from azure to redis; azure table: $azureAccountsTableName, redis: ${config.me.redis.host}.${config.me.redis.port}")
         val wallets = azureDatabaseAccessor.loadWallets()
         val loadTime = Date().time
-        teeLog("Loaded ${wallets.size} wallets from azure (ms: ${loadTime - startTime})")
-        balancesHolder.insertOrUpdateWallets(wallets.values.toList(), null)
+        LOGGER.info("Loaded ${wallets.size} wallets from azure (ms: ${loadTime - startTime})")
+        val balancesSaved = balancesService.insertOrUpdateWallets(wallets.values.toList(), null)
+        if (!balancesSaved) {
+            LOGGER.error("Can not save balances data during migration from azure db to redis")
+            return
+        }
         val saveTime = Date().time
-        teeLog("Saved ${wallets.size} wallets to redis (ms: ${saveTime - loadTime})")
+        LOGGER.info("Saved ${wallets.size} wallets to redis (ms: ${saveTime - loadTime})")
 
         compare()
     }
 
     fun fromRedisToDb() {
         val startTime = Date().time
-        teeLog("Starting wallets migration from redis to azure; redis: ${config.me.redis.host}.${config.me.redis.port}, azure table: $azureAccountsTableName")
+        LOGGER.info("Starting wallets migration from redis to azure; redis: ${config.me.redis.host}.${config.me.redis.port}, azure table: $azureAccountsTableName")
         val loadTime = Date().time
         val wallets = redisWalletDatabaseAccessor.get().loadWallets()
         if (wallets.isEmpty()) {
             throw AccountsMigrationException("There are no wallets in redis ${config.me.redis.host}.${config.me.redis.port}")
         }
-        teeLog("Loaded ${wallets.size} wallets from redis (ms: ${loadTime - startTime})")
-        balancesHolder.insertOrUpdateWallets(wallets.values.toList(), null)
+        LOGGER.info("Loaded ${wallets.size} wallets from redis (ms: ${loadTime - startTime})")
+        val balancesSaved = balancesService.insertOrUpdateWallets(wallets.values.toList(), null)
+        if (!balancesSaved) {
+            LOGGER.error("Can not save balances data during migration from redis to azure db")
+            return
+        }
         val saveTime = Date().time
-        teeLog("Saved ${wallets.size} wallets to azure (ms: ${saveTime - loadTime})")
+        LOGGER.info("Saved ${wallets.size} wallets to azure (ms: ${saveTime - loadTime})")
 
         compare()
     }
 
-    private fun teeLog(message: String) {
-        println(message)
-        LOGGER.info(message)
-    }
-
     /** Compares balances stored in redis & azure; logs comparison result  */
     private fun compare() {
         val azureWallets = azureDatabaseAccessor.loadWallets().filter { it.value.balances.isNotEmpty() }
@@ -98,28 +101,28 @@ class AccountsMigrationService @Autowired constructor (private val balancesHolde
 
         val differentWallets = LinkedList<String>()
 
-        teeLog("Comparison result. Differences: ")
-        teeLog("---------------------------------------------------------------------------------------------")
+        LOGGER.info("Comparison result. Differences: ")
+        LOGGER.info("---------------------------------------------------------------------------------------------")
         commonClients.forEach {
             val azureWallet = azureWallets[it]
             val redisWallet = redisWallets[it]
             if (!compareBalances(azureWallet!!, redisWallet!!)) {
                 differentWallets.add(it)
             }
         }
-        teeLog("---------------------------------------------------------------------------------------------")
-
-        teeLog("Total: ")
-        teeLog("azure clients count: ${azureWallets.size}")
-        teeLog("redis clients count: ${redisWallets.size}")
-        teeLog("only azure clients (count: ${onlyAzureClients.size}): $onlyAzureClients")
-        teeLog("only redis clients (count: ${onlyRedisClients.size}): $onlyRedisClients")
-        teeLog("clients with different wallets (count: ${differentWallets.size}): $differentWallets")
+        LOGGER.info("---------------------------------------------------------------------------------------------")
+
+        LOGGER.info("Total: ")
+        LOGGER.info("azure clients count: ${azureWallets.size}")
+        LOGGER.info("redis clients count: ${redisWallets.size}")
+        LOGGER.info("only azure clients (count: ${onlyAzureClients.size}): $onlyAzureClients")
+        LOGGER.info("only redis clients (count: ${onlyRedisClients.size}): $onlyRedisClients")
+        LOGGER.info("clients with different wallets (count: ${differentWallets.size}): $differentWallets")
     }
 
     private fun compareBalances(azureWallet: Wallet, redisWallet: Wallet): Boolean {
         if (azureWallet.clientId != redisWallet.clientId) {
-            teeLog("different clients: ${azureWallet.clientId} & ${redisWallet.clientId}")
+            LOGGER.info("different clients: ${azureWallet.clientId} & ${redisWallet.clientId}")
             return false
         }
         val clientId = azureWallet.clientId
@@ -130,7 +133,7 @@ class AccountsMigrationService @Autowired constructor (private val balancesHolde
         val onlyRedisAssets = redisBalances.keys.filterNot { azureBalances.keys.contains(it) }
 
         if (onlyAzureAssets.isNotEmpty() || onlyRedisAssets.isNotEmpty()) {
-            teeLog("different asset sets: $onlyAzureAssets & $onlyRedisAssets, client: $clientId")
+            LOGGER.info("different asset sets: $onlyAzureAssets & $onlyRedisAssets, client: $clientId")
             return false
         }
 
@@ -139,11 +142,11 @@ class AccountsMigrationService @Autowired constructor (private val balancesHolde
             val azureBalance = azureBalances[it]
             val redisBalance = redisBalances[it]
             if (azureBalance!!.balance != redisBalance!!.balance) {
-                teeLog("different balances: ${azureBalance.balance} & ${redisBalance.balance}, client: $clientId")
+                LOGGER.info("different balances: ${azureBalance.balance} & ${redisBalance.balance}, client: $clientId")
                 return false
             }
             if (azureBalance.reserved != redisBalance.reserved) {
-                teeLog("different reserved balances: ${azureBalance.reserved} & ${redisBalance.reserved}, client: $clientId")
+                LOGGER.info("different reserved balances: ${azureBalance.reserved} & ${redisBalance.reserved}, client: $clientId")
                 return false
             }
         }

src/main/kotlin/com/lykke/matching/engine/database/reconciliation/OrdersMigrationService.kt

@@ -45,7 +45,7 @@ class OrdersMigrationService(private val config: Config,
             return
         }
         if (config.me.storage != Storage.Redis) {
-            teeLog("Do not perform migration to files")
+            LOGGER.info("Do not perform migration to files")
             return
         }
         fromFilesToRedis()
@@ -59,12 +59,12 @@ class OrdersMigrationService(private val config: Config,
             throw Exception("Stop orders already exist in redis ${config.me.redis.host}.${config.me.redis.port}/${config.me.redis.ordersDatabase}")
         }
         val startTime = Date().time
-        teeLog("Starting orders migration from files to redis; files dirs: ${config.me.orderBookPath}, ${config.me.stopOrderBookPath}" +
+        LOGGER.info("Starting orders migration from files to redis; files dirs: ${config.me.orderBookPath}, ${config.me.stopOrderBookPath}" +
                 ", redis: ${config.me.redis.host}.${config.me.redis.port}/${config.me.redis.ordersDatabase}")
         val orders = fileOrderBookDatabaseAccessor.loadLimitOrders()
         val stopOrders = fileStopOrderBookDatabaseAccessor.loadStopLimitOrders()
         val loadTime = Date().time
-        teeLog("Loaded ${orders.size} orders from files (ms: ${loadTime - startTime})")
+        LOGGER.info("Loaded ${orders.size} orders from files (ms: ${loadTime - startTime})")
         persistenceManager.persist(PersistenceData(null,
                 null,
                 OrderBooksPersistenceData(mapOrdersToOrderBookPersistenceDataList(orders, LOGGER),
@@ -77,11 +77,6 @@ class OrdersMigrationService(private val config: Config,
         genericLimitOrderService.update()
         genericStopLimitOrderService.update()
         val saveTime = Date().time
-        teeLog("Saved ${orders.size} orders and ${stopOrders.size} stop orders to redis (ms: ${saveTime - loadTime})")
-    }
-
-    private fun teeLog(message: String) {
-        println(message)
-        LOGGER.info(message)
+        LOGGER.info("Saved ${orders.size} orders and ${stopOrders.size} stop orders to redis (ms: ${saveTime - loadTime})")
     }
 }