From 06302e5394d332027a7a5361d687536155dc0840 Mon Sep 17 00:00:00 2001 From: Artur Albov Date: Sun, 22 Apr 2018 18:38:03 +0300 Subject: [PATCH 1/2] #82 Contract Summary Tx processing stucks --- CHANGELOG.md | 2 + .../UpdateContractSummaryRepository.kt | 3 +- .../UpdateContractSummaryRepository.kt | 3 +- .../apply/BitcoinTxConsumerConfiguration.kt | 6 +- .../apply/UpdateContractSummaryProcess.kt | 58 +++++++++++-------- .../apply/EthereumConsumerConfiguration.kt | 28 +++++---- 6 files changed, 60 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index db1534b0..a9d07453 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,11 +8,13 @@ and this project adheres to [Semantic Visioning](http://semver.org/spec/v2.0.0.h ## [0.3.0] - Bitcoin pump, dump, contract summary ### Added +- [#108](/../../issues/108) Create API for search results preview - [#106](/../../issues/106) Add filter params for search api - [#84](/../../issues/84) API: bitcoin endpoints - [#98](/../../issues/98) Make chain items retrivial caps non-depended - [#85](/../../issues/85) Bitcoin pump/dump and bitcoin contract summary docker images, dockerhub ### Fixed +- [#113](/../../issues/113) Make search API logic dependent of existing keyspaces - [#99](/../../issues/99) Pump stuck if chain reorganization bundles number exceed history stack size diff --git a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt index 46ced3ab..bdff8037 100644 --- a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt +++ b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt @@ -19,7 +19,8 @@ interface BitcoinUpdateContractSummaryRepository : ReactiveCrudRepository @Consistency(value = ConsistencyLevel.LOCAL_QUORUM) - fun findAllByHashIn(hashes: Iterable): Flux + fun findAllByHashIn(hashes: Iterable): Flux = Flux.fromIterable(hashes) + .flatMap { hash -> findByHash(hash) } /** * Return {@code true} if update was successful. diff --git a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt index f32eaf78..2b50bc3b 100644 --- a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt +++ b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt @@ -20,7 +20,8 @@ interface EthereumUpdateContractSummaryRepository : ReactiveCrudRepository @Consistency(value = ConsistencyLevel.LOCAL_QUORUM) - fun findAllByHashIn(hashes: Iterable): Flux + fun findAllByHashIn(hashes: Iterable): Flux = Flux.fromIterable(hashes) + .flatMap { hash -> findByHash(hash) } /** * Return {@code true} if update was successful. diff --git a/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt b/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt index d1c9c589..556d6124 100644 --- a/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt +++ b/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt @@ -29,6 +29,8 @@ import org.springframework.kafka.listener.config.ContainerProperties import org.springframework.transaction.annotation.EnableTransactionManagement private const val MAX_POLL_RECORDS_CONFIG = 500 +private const val MAX_POLL_INTERVAL_MS_CONFIG = 20000 +private const val SESSION_TIMEOUT_MS_CONFIG = 30000 @EnableKafka @Configuration @@ -79,6 +81,8 @@ class BitcoinTxConsumerConfiguration { ConsumerConfig.GROUP_ID_CONFIG to "bitcoin-contract-summary-update-process", ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, ConsumerConfig.ISOLATION_LEVEL_CONFIG to IsolationLevel.READ_COMMITTED.toString().toLowerCase(), - ConsumerConfig.MAX_POLL_RECORDS_CONFIG to MAX_POLL_RECORDS_CONFIG + ConsumerConfig.MAX_POLL_RECORDS_CONFIG to MAX_POLL_RECORDS_CONFIG, + ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to MAX_POLL_INTERVAL_MS_CONFIG, + ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG to SESSION_TIMEOUT_MS_CONFIG ) } diff --git a/contract-summary/common/src/main/kotlin/fund/cyber/contract/common/delta/apply/UpdateContractSummaryProcess.kt b/contract-summary/common/src/main/kotlin/fund/cyber/contract/common/delta/apply/UpdateContractSummaryProcess.kt index c56156fb..89d3007b 100644 --- a/contract-summary/common/src/main/kotlin/fund/cyber/contract/common/delta/apply/UpdateContractSummaryProcess.kt +++ b/contract-summary/common/src/main/kotlin/fund/cyber/contract/common/delta/apply/UpdateContractSummaryProcess.kt @@ -31,13 +31,13 @@ fun Flux.await(): List { } fun CqlContractSummary.hasSameTopicPartitionAs(delta: ContractSummaryDelta<*>) = - this.kafkaDeltaTopic == delta.topic && this.kafkaDeltaPartition == delta.partition + this.kafkaDeltaTopic == delta.topic && this.kafkaDeltaPartition == delta.partition fun CqlContractSummary.hasSameTopicPartitionAs(topic: String, partition: Int) = - this.kafkaDeltaTopic == topic && this.kafkaDeltaPartition == partition + this.kafkaDeltaTopic == topic && this.kafkaDeltaPartition == partition fun CqlContractSummary.notSameTopicPartitionAs(delta: ContractSummaryDelta<*>) = - hasSameTopicPartitionAs(delta).not() + hasSameTopicPartitionAs(delta).not() fun CqlContractSummary.committed() = this.kafkaDeltaOffsetCommitted @@ -51,14 +51,14 @@ private const val MAX_STORE_ATTEMPTS = 20 private const val STORE_RETRY_TIMEOUT = 30L data class UpdateInfo( - val topic: String, - val partition: Int, - val minOffset: Long, - val maxOffset: Long + val topic: String, + val partition: Int, + val minOffset: Long, + val maxOffset: Long ) { constructor(records: List>) : this( - topic = records.first().topic(), partition = records.first().partition(), - minOffset = records.first().offset(), maxOffset = records.last().offset() + topic = records.first().topic(), partition = records.first().partition(), + minOffset = records.first().offset(), maxOffset = records.last().offset() ) } @@ -83,29 +83,34 @@ class UpdateContractSummaryProcess>, consumer: Consumer<*, *>) { - val info = UpdateInfo(records.sortedBy { record -> record.offset() }) + initMonitors(info) + recordsProcessingTimer.recordCallable { processRecords(records, consumer, info) } + } + + private fun processRecords(records: List>, consumer: Consumer<*, *>, + info: UpdateInfo) { log.info("Processing records for topic: ${info.topic}; partition ${info.partition} from ${info.minOffset}" + - " to ${info.maxOffset} offset.") + " to ${info.maxOffset} offset.") val storeAttempts: MutableMap = mutableMapOf() val previousStates: MutableMap = mutableMapOf() - initMonitors(info) val contracts = deltaProcessor.affectedContracts(records) val contractsSummary = contractSummaryStorage.findAllByIdIn(contracts) - .await().groupBy { a -> a.hash }.map { (k, v) -> k to v.first() }.toMap() + .await().groupBy { a -> a.hash }.map { (k, v) -> k to v.first() }.toMap() val deltas = records.flatMap { record -> deltaProcessor.recordToDeltas(record) } val mergedDeltas = deltas.groupBy { delta -> delta.contract } - .filterKeys { contract -> contract.isNotEmpty() } - .mapValues { contractDeltas -> deltaMerger.mergeDeltas(contractDeltas.value, contractsSummary) } - .filterValues { value -> value != null } - .map { entry -> entry.key to entry.value!! }.toMap() + .filterKeys { contract -> contract.isNotEmpty() } + .mapValues { contractDeltas -> deltaMerger.mergeDeltas(contractDeltas.value, contractsSummary) } + .filterValues { value -> value != null } + .map { entry -> entry.key to entry.value!! }.toMap() try { @@ -140,11 +145,11 @@ class UpdateContractSummaryProcess if (summary.notCommitted() && summary.hasSameTopicPartitionAs(info.topic, info.partition) - && summary.kafkaDeltaOffset in info.minOffset..info.maxOffset) { + && summary.kafkaDeltaOffset in info.minOffset..info.maxOffset) { val previousState = previousStates[summary.hash] if (previousState != null) { contractSummaryStorage.update(previousState) @@ -208,7 +213,7 @@ class UpdateContractSummaryProcess this.kafkaDeltaOffset//todo : = or >= ???? + lastOffsetOf(this.kafkaDeltaTopic, this.kafkaDeltaPartition) > this.kafkaDeltaOffset//todo : = or >= ???? private fun lastOffsetOf(topic: String, partition: Int): Long { val reader = SinglePartitionTopicLastItemsReader( - kafkaBrokers = kafkaBrokers, topic = topic, - keyClass = Any::class.java, valueClass = Any::class.java + kafkaBrokers = kafkaBrokers, topic = topic, + keyClass = Any::class.java, valueClass = Any::class.java ) return reader.readLastOffset(partition) } diff --git a/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt b/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt index c3355794..fdf185e6 100644 --- a/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt +++ b/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt @@ -35,6 +35,8 @@ import org.springframework.kafka.listener.config.ContainerProperties import org.springframework.transaction.annotation.EnableTransactionManagement private const val MAX_POLL_RECORDS_CONFIG = 500 +private const val MAX_POLL_INTERVAL_MS_CONFIG = 20000 +private const val SESSION_TIMEOUT_MS_CONFIG = 30000 @EnableKafka @Configuration @@ -71,13 +73,13 @@ class EthereumTxConsumerConfiguration { fun txListenerContainer(): ConcurrentMessageListenerContainer { val consumerFactory = DefaultKafkaConsumerFactory( - consumerConfigs(), JsonDeserializer(PumpEvent::class.java), JsonDeserializer(EthereumTx::class.java) + consumerConfigs(), JsonDeserializer(PumpEvent::class.java), JsonDeserializer(EthereumTx::class.java) ) val containerProperties = ContainerProperties(chain.txPumpTopic).apply { setBatchErrorHandler(SeekToCurrentBatchErrorHandler()) messageListener = UpdateContractSummaryProcess(contractSummaryStorage, txDeltaProcessor, deltaMerger, - monitoring, kafkaBrokers) + monitoring, kafkaBrokers) isAckOnError = false ackMode = AbstractMessageListenerContainer.AckMode.BATCH } @@ -91,13 +93,13 @@ class EthereumTxConsumerConfiguration { fun blockListenerContainer(): ConcurrentMessageListenerContainer { val consumerFactory = DefaultKafkaConsumerFactory( - consumerConfigs(), JsonDeserializer(PumpEvent::class.java), JsonDeserializer(EthereumBlock::class.java) + consumerConfigs(), JsonDeserializer(PumpEvent::class.java), JsonDeserializer(EthereumBlock::class.java) ) val containerProperties = ContainerProperties(chain.blockPumpTopic).apply { setBatchErrorHandler(SeekToCurrentBatchErrorHandler()) messageListener = UpdateContractSummaryProcess(contractSummaryStorage, blockDeltaProcessor, deltaMerger, - monitoring, kafkaBrokers) + monitoring, kafkaBrokers) isAckOnError = false ackMode = AbstractMessageListenerContainer.AckMode.BATCH } @@ -111,13 +113,13 @@ class EthereumTxConsumerConfiguration { fun uncleListenerContainer(): ConcurrentMessageListenerContainer { val consumerFactory = DefaultKafkaConsumerFactory( - consumerConfigs(), JsonDeserializer(PumpEvent::class.java), JsonDeserializer(EthereumUncle::class.java) + consumerConfigs(), JsonDeserializer(PumpEvent::class.java), JsonDeserializer(EthereumUncle::class.java) ) val containerProperties = ContainerProperties(chain.unclePumpTopic).apply { setBatchErrorHandler(SeekToCurrentBatchErrorHandler()) messageListener = UpdateContractSummaryProcess(contractSummaryStorage, uncleDeltaProcessor, deltaMerger, - monitoring, kafkaBrokers) + monitoring, kafkaBrokers) isAckOnError = false ackMode = AbstractMessageListenerContainer.AckMode.BATCH } @@ -128,11 +130,13 @@ class EthereumTxConsumerConfiguration { } private fun consumerConfigs(): MutableMap = defaultConsumerConfig().with( - ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers, - ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest", - ConsumerConfig.GROUP_ID_CONFIG to "ethereum-contract-summary-update-process", - ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, - ConsumerConfig.ISOLATION_LEVEL_CONFIG to IsolationLevel.READ_COMMITTED.toString().toLowerCase(), - ConsumerConfig.MAX_POLL_RECORDS_CONFIG to MAX_POLL_RECORDS_CONFIG + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers, + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest", + ConsumerConfig.GROUP_ID_CONFIG to "ethereum-contract-summary-update-process", + ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, + ConsumerConfig.ISOLATION_LEVEL_CONFIG to IsolationLevel.READ_COMMITTED.toString().toLowerCase(), + ConsumerConfig.MAX_POLL_RECORDS_CONFIG to MAX_POLL_RECORDS_CONFIG, + ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to MAX_POLL_INTERVAL_MS_CONFIG, + ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG to SESSION_TIMEOUT_MS_CONFIG ) } From 1cfe67b6cc2cade814d1e67e116b53663121c1bd Mon Sep 17 00:00:00 2001 From: Artur Albov Date: Mon, 23 Apr 2018 10:59:37 +0300 Subject: [PATCH 2/2] #82 Contract Summary Tx processing stucks --- .../bitcoin/repository/UpdateContractSummaryRepository.kt | 2 +- .../ethereum/repository/UpdateContractSummaryRepository.kt | 2 +- .../cyber/contract/bitcoin/BitcoinContractSummaryStorage.kt | 2 +- .../bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt | 2 -- .../cyber/contract/ethereum/EthereumContractSummaryStorage.kt | 2 +- .../ethereum/delta/apply/EthereumConsumerConfiguration.kt | 2 -- 6 files changed, 4 insertions(+), 8 deletions(-) diff --git a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt index bdff8037..98263ec8 100644 --- a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt +++ b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/bitcoin/repository/UpdateContractSummaryRepository.kt @@ -19,7 +19,7 @@ interface BitcoinUpdateContractSummaryRepository : ReactiveCrudRepository @Consistency(value = ConsistencyLevel.LOCAL_QUORUM) - fun findAllByHashIn(hashes: Iterable): Flux = Flux.fromIterable(hashes) + fun findAllByHash(hashes: Iterable): Flux = Flux.fromIterable(hashes) .flatMap { hash -> findByHash(hash) } /** diff --git a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt index 2b50bc3b..cfe6f44e 100644 --- a/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt +++ b/cassandra-service/src/main/kotlin/fund/cyber/cassandra/ethereum/repository/UpdateContractSummaryRepository.kt @@ -20,7 +20,7 @@ interface EthereumUpdateContractSummaryRepository : ReactiveCrudRepository @Consistency(value = ConsistencyLevel.LOCAL_QUORUM) - fun findAllByHashIn(hashes: Iterable): Flux = Flux.fromIterable(hashes) + fun findAllByHash(hashes: Iterable): Flux = Flux.fromIterable(hashes) .flatMap { hash -> findByHash(hash) } /** diff --git a/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/BitcoinContractSummaryStorage.kt b/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/BitcoinContractSummaryStorage.kt index 91ab1550..080ecec6 100644 --- a/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/BitcoinContractSummaryStorage.kt +++ b/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/BitcoinContractSummaryStorage.kt @@ -15,7 +15,7 @@ class BitcoinContractSummaryStorage( override fun findById(id: String): Mono = contractSummaryRepository.findByHash(id) override fun findAllByIdIn(ids: Iterable): Flux = contractSummaryRepository - .findAllByHashIn(ids) + .findAllByHash(ids) override fun update(summary: CqlBitcoinContractSummary, oldVersion: Long): Mono = contractSummaryRepository .update(summary, oldVersion) diff --git a/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt b/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt index 556d6124..60780e5a 100644 --- a/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt +++ b/contract-summary/bitcoin/src/main/kotlin/fund/cyber/contract/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt @@ -29,7 +29,6 @@ import org.springframework.kafka.listener.config.ContainerProperties import org.springframework.transaction.annotation.EnableTransactionManagement private const val MAX_POLL_RECORDS_CONFIG = 500 -private const val MAX_POLL_INTERVAL_MS_CONFIG = 20000 private const val SESSION_TIMEOUT_MS_CONFIG = 30000 @EnableKafka @@ -82,7 +81,6 @@ class BitcoinTxConsumerConfiguration { ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, ConsumerConfig.ISOLATION_LEVEL_CONFIG to IsolationLevel.READ_COMMITTED.toString().toLowerCase(), ConsumerConfig.MAX_POLL_RECORDS_CONFIG to MAX_POLL_RECORDS_CONFIG, - ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to MAX_POLL_INTERVAL_MS_CONFIG, ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG to SESSION_TIMEOUT_MS_CONFIG ) } diff --git a/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/EthereumContractSummaryStorage.kt b/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/EthereumContractSummaryStorage.kt index cd1e2f64..05a7d287 100644 --- a/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/EthereumContractSummaryStorage.kt +++ b/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/EthereumContractSummaryStorage.kt @@ -15,7 +15,7 @@ class EthereumContractSummaryStorage( override fun findById(id: String): Mono = contractSummaryRepository.findByHash(id) override fun findAllByIdIn(ids: Iterable): Flux = contractSummaryRepository - .findAllByHashIn(ids) + .findAllByHash(ids) override fun update(summary: CqlEthereumContractSummary, oldVersion: Long): Mono = contractSummaryRepository.update(summary, oldVersion) diff --git a/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt b/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt index fdf185e6..359e7f8e 100644 --- a/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt +++ b/contract-summary/ethereum/src/main/kotlin/fund/cyber/contract/ethereum/delta/apply/EthereumConsumerConfiguration.kt @@ -35,7 +35,6 @@ import org.springframework.kafka.listener.config.ContainerProperties import org.springframework.transaction.annotation.EnableTransactionManagement private const val MAX_POLL_RECORDS_CONFIG = 500 -private const val MAX_POLL_INTERVAL_MS_CONFIG = 20000 private const val SESSION_TIMEOUT_MS_CONFIG = 30000 @EnableKafka @@ -136,7 +135,6 @@ class EthereumTxConsumerConfiguration { ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, ConsumerConfig.ISOLATION_LEVEL_CONFIG to IsolationLevel.READ_COMMITTED.toString().toLowerCase(), ConsumerConfig.MAX_POLL_RECORDS_CONFIG to MAX_POLL_RECORDS_CONFIG, - ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to MAX_POLL_INTERVAL_MS_CONFIG, ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG to SESSION_TIMEOUT_MS_CONFIG ) }