Dev features and enhancements

Jenkinsfile

@@ -8,7 +8,7 @@ pipeline {
                 withMaven(
                         maven: 'maven-3.6.3'
                 ) {
-                    sh 'mvn -B clean install'
+                    sh 'mvn -T 1C -B clean install'
                 }
             }
         }
@@ -20,8 +20,14 @@ pipeline {
                 SMTP_PASS = credentials("smtp-secret")
                 DB_USER = 'opex'
                 DB_PASS = credentials("db-secret")
+                DB_BACKUP_USER = 'opex_backup'
+                DB_BACKUP_PASS = credentials("db-backup-secret")
                 KEYCLOAK_ADMIN_URL = 'https://demo.opex.dev/auth'
                 KEYCLOAK_FRONTEND_URL = 'https://demo.opex.dev/auth'
+                KEYCLOAK_ADMIN_USERNAME = credentials("keycloak-admin-username")
+                KEYCLOAK_ADMIN_PASSWORD = credentials("keycloak-admin-password")
+                OPEX_ADMIN_KEYCLOAK_CLIENT_SECRET = credentials("opex-admin-keycloak-client-secret")
+                VANDAR_API_KEY = credentials("vandar-api-key")
                 COMPOSE_PROJECT_NAME = 'demo-core'
                 DEFAULT_NETWORK_NAME = 'demo-opex'
             }

accountant/accountant-app/src/main/resources/application.yml

@@ -13,9 +13,6 @@ spring:
     bootstrap-servers: ${KAFKA_IP_PORT:localhost:9092}
     consumer:
       group-id: accountant
-  redis:
-    hostname: ${REDIS_HOST:localhost}
-    port: 6379
   r2dbc:
     url: r2dbc:postgresql://${DB_IP_PORT:localhost}/opex_accountant
     username: ${dbusername:opex}

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/model/PairConfig.kt

@@ -5,5 +5,6 @@ class PairConfig(
     val leftSideWalletSymbol: String, //can be same as pair left side
     val rightSideWalletSymbol: String, //can be same as pair right side
     val leftSideFraction: Double,
-    val rightSideFraction: Double
+    val rightSideFraction: Double,
+    val rate: Double = 0.0
 )

accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/config/AccountantKafkaConfig.kt

@@ -1,39 +1,38 @@
 package co.nilin.opex.accountant.ports.kafka.listener.config
 
-
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.EventKafkaListener
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.OrderKafkaListener
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.TempEventKafkaListener
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.TradeKafkaListener
 import co.nilin.opex.matching.engine.core.eventh.events.CoreEvent
-import org.apache.kafka.clients.admin.NewTopic
 import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
-import org.springframework.context.support.GenericApplicationContext
 import org.springframework.kafka.core.ConsumerFactory
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory
-import org.springframework.kafka.listener.ConcurrentMessageListenerContainer
-import org.springframework.kafka.listener.ContainerProperties
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.listener.*
 import org.springframework.kafka.support.serializer.JsonDeserializer
+import org.springframework.util.backoff.FixedBackOff
 import java.util.regex.Pattern
 
 @Configuration
 class AccountantKafkaConfig {
 
     @Value("\${spring.kafka.bootstrap-servers}")
-    private val bootstrapServers: String? = null
+    private lateinit var bootstrapServers: String
 
     @Value("\${spring.kafka.consumer.group-id}")
-    private val groupId: String? = null
+    private lateinit var groupId: String
 
-    @Bean("accountantConsumerConfig")
-    fun consumerConfigs(): Map<String, Any?> {
+    @Bean("consumerConfig")
+    fun consumerConfigs(): Map<String, Any> {
         return mapOf(
             ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
             ConsumerConfig.GROUP_ID_CONFIG to groupId,
@@ -45,66 +44,76 @@ class AccountantKafkaConfig {
     }
 
     @Bean("accountantConsumerFactory")
-    fun consumerFactory(@Qualifier("accountantConsumerConfig") consumerConfigs: Map<String, Any?>): ConsumerFactory<String, CoreEvent> {
+    fun consumerFactory(@Qualifier("consumerConfig") consumerConfigs: Map<String, Any?>): ConsumerFactory<String, CoreEvent> {
         return DefaultKafkaConsumerFactory(consumerConfigs)
     }
 
     @Autowired
     @ConditionalOnBean(TradeKafkaListener::class)
     fun configureTradeListener(
         tradeListener: TradeKafkaListener,
+        @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate<String?, CoreEvent>,
         @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory<String, CoreEvent>
     ) {
         val containerProps = ContainerProperties(Pattern.compile("trades_.*"))
         containerProps.messageListener = tradeListener
         val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps)
         container.setBeanName("TradeKafkaListenerContainer")
+        container.commonErrorHandler = createConsumerErrorHandler(template, "trades.DLT")
         container.start()
     }
 
     @Autowired
     @ConditionalOnBean(EventKafkaListener::class)
     fun configureEventListener(
         eventListener: EventKafkaListener,
+        @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate<String?, CoreEvent>,
         @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory<String, CoreEvent>
     ) {
         val containerProps = ContainerProperties(Pattern.compile("events_.*"))
         containerProps.messageListener = eventListener
         val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps)
         container.setBeanName("EventKafkaListenerContainer")
+        container.commonErrorHandler = createConsumerErrorHandler(template, "events.DLT")
         container.start()
     }
 
     @Autowired
     @ConditionalOnBean(OrderKafkaListener::class)
     fun configureOrderListener(
         orderListener: OrderKafkaListener,
+        @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate<String?, CoreEvent>,
         @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory<String, CoreEvent>
     ) {
         val containerProps = ContainerProperties(Pattern.compile("orders_.*"))
         containerProps.messageListener = orderListener
         val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps)
         container.setBeanName("OrderKafkaListenerContainer")
+        container.commonErrorHandler = createConsumerErrorHandler(template, "orders.DLT")
         container.start()
     }
 
-    @Autowired
-    fun createTempTopics(applicationContext: GenericApplicationContext) {
-        applicationContext.registerBean("topic_tempevents", NewTopic::class.java, "tempevents", 1, 1)
-    }
-
     @Autowired
     @ConditionalOnBean(TempEventKafkaListener::class)
     fun configureTempEventListener(
         eventListener: TempEventKafkaListener,
+        @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate<String?, CoreEvent>,
         @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory<String, CoreEvent>
     ) {
         val containerProps = ContainerProperties(Pattern.compile("tempevents"))
         containerProps.messageListener = eventListener
         val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps)
         container.setBeanName("TempEventKafkaListenerContainer")
+        container.commonErrorHandler = createConsumerErrorHandler(template, "tempevents.DLT")
         container.start()
     }
 
+    private fun createConsumerErrorHandler(kafkaTemplate: KafkaTemplate<*, *>, dltTopic: String): CommonErrorHandler {
+        val recoverer = DeadLetterPublishingRecoverer(kafkaTemplate) { cr, _ ->
+            cr.headers().add("dlt-origin-module", "ACCOUNTANT".toByteArray())
+            TopicPartition(dltTopic, cr.partition())
+        }
+        return DefaultErrorHandler(recoverer, FixedBackOff(5_000, 20))
+    }
 
 }

accountant/accountant-ports/accountant-persister-postgres/pom.xml

@@ -68,12 +68,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-
-    <repositories>
-        <repository>
-            <id>spring-milestones</id>
-            <name>Spring Milestones</name>
-            <url>https://repo.spring.io/milestone</url>
-        </repository>
-    </repositories>
 </project>

accountant/accountant-ports/accountant-persister-postgres/src/main/resources/data.sql

@@ -5,6 +5,14 @@ VALUES ('btc_usdt', 'btc', 'usdt', 0.000001, 0.01, 55000),
        ('nln_btc', 'nln', 'btc', 1.0, 0.000001, 1 / 5500000)
 ON CONFLICT DO NOTHING;
 
+-- Test pairs
+INSERT INTO pair_config
+VALUES ('tbtc_tusdt', 'tbtc', 'tusdt', 0.000001, 0.01, 55000),
+       ('teth_tusdt', 'teth', 'tusdt', 0.00001, 0.01, 3800),
+       ('nln_tusdt', 'nln', 'tusdt', 1.0, 0.01, 0.01),
+       ('nln_tbtc', 'nln', 'tbtc', 1.0, 0.000001, 1 / 5500000)
+ON CONFLICT DO NOTHING;
+
 INSERT INTO pair_fee_config
 VALUES (1, 'btc_usdt', 'ASK', '*', 0.01, 0.01),
        (2, 'btc_usdt', 'BID', '*', 0.01, 0.01),
@@ -16,6 +24,18 @@ VALUES (1, 'btc_usdt', 'ASK', '*', 0.01, 0.01),
        (8, 'eth_usdt', 'BID', '*', 0.01, 0.01)
 ON CONFLICT DO NOTHING;
 
+-- Test pair configs
+INSERT INTO pair_fee_config
+VALUES (9, 'tbtc_tusdt', 'ASK', '*', 0.01, 0.01),
+       (10, 'tbtc_tusdt', 'BID', '*', 0.01, 0.01),
+       (11, 'nln_tusdt', 'ASK', '*', 0.01, 0.01),
+       (12, 'nln_tusdt', 'BID', '*', 0.01, 0.01),
+       (13, 'nln_tbtc', 'ASK', '*', 0.01, 0.01),
+       (14, 'nln_tbtc', 'BID', '*', 0.01, 0.01),
+       (15, 'teth_tusdt', 'ASK', '*', 0.01, 0.01),
+       (16, 'teth_tusdt', 'BID', '*', 0.01, 0.01)
+ON CONFLICT DO NOTHING;
+
 SELECT setval(pg_get_serial_sequence('pair_fee_config', 'id'), (SELECT MAX(id) FROM pair_fee_config));
 
 COMMIT;

accountant/accountant-ports/accountant-submitter-kafka/pom.xml

@@ -58,12 +58,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-
-    <repositories>
-        <repository>
-            <id>spring-milestones</id>
-            <name>Spring Milestones</name>
-            <url>https://repo.spring.io/milestone</url>
-        </repository>
-    </repositories>
 </project>

accountant/accountant-ports/accountant-submitter-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/submitter/config/KafkaTopicConfig.kt

@@ -0,0 +1,37 @@
+package co.nilin.opex.accountant.ports.kafka.submitter.config
+
+import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.common.config.TopicConfig
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.support.GenericApplicationContext
+import org.springframework.kafka.config.TopicBuilder
+import java.util.function.Supplier
+
+@Configuration
+class KafkaTopicConfig {
+
+    @Autowired
+    fun createTopics(applicationContext: GenericApplicationContext) {
+        with(applicationContext) {
+            registerBean("topic_richOrder", NewTopic::class.java, Supplier {
+                TopicBuilder.name("richOrder")
+                    .partitions(10)
+                    .replicas(3)
+                    .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
+                    .build()
+            })
+
+            registerBean("topic_richTrade", NewTopic::class.java, Supplier {
+                TopicBuilder.name("richTrade")
+                    .partitions(10)
+                    .replicas(3)
+                    .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
+                    .build()
+            })
+
+            registerBean("topic_tempevents", NewTopic::class.java, "tempevents", 1, 1)
+        }
+    }
+
+}

accountant/accountant-ports/accountant-submitter-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/submitter/config/SubmitterKafkaConfig.kt

@@ -3,30 +3,24 @@ package co.nilin.opex.accountant.ports.kafka.submitter.config
 import co.nilin.opex.accountant.core.inout.RichOrderEvent
 import co.nilin.opex.accountant.core.inout.RichTrade
 import co.nilin.opex.matching.engine.core.eventh.events.CoreEvent
-import org.apache.kafka.clients.admin.NewTopic
 import org.apache.kafka.clients.producer.ProducerConfig
-import org.apache.kafka.common.config.TopicConfig
 import org.apache.kafka.common.serialization.StringSerializer
-import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
-import org.springframework.context.support.GenericApplicationContext
-import org.springframework.kafka.config.TopicBuilder
 import org.springframework.kafka.core.DefaultKafkaProducerFactory
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.core.ProducerFactory
 import org.springframework.kafka.support.serializer.JsonSerializer
-import java.util.function.Supplier
 
 @Configuration
 class SubmitterKafkaConfig {
 
     @Value("\${spring.kafka.bootstrap-servers}")
     private lateinit var bootstrapServers: String
 
-    @Bean("accountantProducerConfigs")
+    @Bean("producerConfigs")
     fun producerConfigs(): Map<String, Any> {
         return mapOf(
             ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
@@ -38,7 +32,7 @@ class SubmitterKafkaConfig {
     }
 
     @Bean("accountantEventProducerFactory")
-    fun producerFactory(@Qualifier("accountantProducerConfigs") producerConfigs: Map<String, Any>): ProducerFactory<String?, CoreEvent> {
+    fun producerFactory(@Qualifier("producerConfigs") producerConfigs: Map<String, Any>): ProducerFactory<String?, CoreEvent> {
         return DefaultKafkaProducerFactory(producerConfigs)
     }
 
@@ -48,7 +42,7 @@ class SubmitterKafkaConfig {
     }
 
     @Bean("richTradeProducerFactory")
-    fun richTradeProducerFactory(@Qualifier("accountantProducerConfigs") producerConfigs: Map<String, Any>): ProducerFactory<String?, RichTrade> {
+    fun richTradeProducerFactory(@Qualifier("producerConfigs") producerConfigs: Map<String, Any>): ProducerFactory<String?, RichTrade> {
         return DefaultKafkaProducerFactory(producerConfigs)
     }
 
@@ -58,31 +52,12 @@ class SubmitterKafkaConfig {
     }
 
     @Bean("richOrderProducerFactory")
-    fun richOrderProducerFactory(@Qualifier("accountantProducerConfigs") producerConfigs: Map<String, Any>): ProducerFactory<String?, RichOrderEvent> {
+    fun richOrderProducerFactory(@Qualifier("producerConfigs") producerConfigs: Map<String, Any>): ProducerFactory<String?, RichOrderEvent> {
         return DefaultKafkaProducerFactory(producerConfigs)
     }
 
     @Bean("richOrderKafkaTemplate")
     fun richOrderKafkaTemplate(@Qualifier("richOrderProducerFactory") producerFactory: ProducerFactory<String?, RichOrderEvent>): KafkaTemplate<String?, RichOrderEvent> {
         return KafkaTemplate(producerFactory)
     }
-
-    @Autowired
-    fun createTopics(applicationContext: GenericApplicationContext) {
-        applicationContext.registerBean("topic_richOrder", NewTopic::class.java, Supplier {
-            TopicBuilder.name("richOrder")
-                .partitions(10)
-                .replicas(3)
-                .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
-                .build()
-        })
-
-        applicationContext.registerBean("topic_richTrade", NewTopic::class.java, Supplier {
-            TopicBuilder.name("richTrade")
-                .partitions(10)
-                .replicas(3)
-                .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
-                .build()
-        })
-    }
 }

accountant/accountant-ports/accountant-submitter-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/submitter/service/RichOrderSubmitter.kt

@@ -2,6 +2,7 @@ package co.nilin.opex.accountant.ports.kafka.submitter.service
 
 import co.nilin.opex.accountant.core.inout.RichOrderEvent
 import co.nilin.opex.accountant.core.spi.RichOrderPublisher
+import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.stereotype.Component
@@ -13,13 +14,17 @@ import kotlin.coroutines.suspendCoroutine
 class RichOrderSubmitter(@Qualifier("richOrderKafkaTemplate") val kafkaTemplate: KafkaTemplate<String, RichOrderEvent>) :
     RichOrderPublisher {
 
+    private val logger = LoggerFactory.getLogger(RichOrderSubmitter::class.java)
+
     override suspend fun publish(order: RichOrderEvent): Unit = suspendCoroutine { cont ->
-        println("richOrderSubmit!")
+        logger.info("Submitting RichOrder")
+
         val sendFuture = kafkaTemplate.send("richOrder", order)
-        sendFuture.addCallback({ sendResult ->
+        sendFuture.addCallback({
             cont.resume(Unit)
-        }, { exception ->
-            cont.resumeWithException(exception)
+        }, {
+            logger.info("Error submitting RichOrder", it)
+            cont.resumeWithException(it)
         })
     }
 }