#32 Error sending message to kafka

address-summary/bitcoin/src/main/kotlin/fund/cyber/address/bitcoin/delta/apply/BitcoinTxConsumerConfiguration.kt

@@ -5,6 +5,8 @@ import fund.cyber.common.kafka.JsonDeserializer
 import fund.cyber.address.bitcoin.BitcoinAddressSummaryStorage
 import fund.cyber.address.bitcoin.summary.BitcoinDeltaMerger
 import fund.cyber.address.bitcoin.summary.BitcoinTxDeltaProcessor
+import fund.cyber.common.kafka.defaultConsumerConfig
+import fund.cyber.common.with
 import fund.cyber.search.configuration.KAFKA_BROKERS
 import fund.cyber.search.configuration.KAFKA_BROKERS_DEFAULT
 import fund.cyber.search.model.bitcoin.BitcoinTx
@@ -71,7 +73,7 @@ class BitcoinTxConsumerConfiguration {
         }
     }
 
-    private fun consumerConfigs(): MutableMap<String, Any> = mutableMapOf(
+    private fun consumerConfigs(): MutableMap<String, Any> = defaultConsumerConfig().with(
             ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers,
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
             ConsumerConfig.GROUP_ID_CONFIG to "bitcoin-address-summary-update-process",

address-summary/ethereum/src/main/kotlin/fund/cyber/address/ethereum/delta/apply/EthereumConsumerConfiguration.kt

@@ -7,6 +7,8 @@ import fund.cyber.address.ethereum.summary.EthereumDeltaMerger
 import fund.cyber.address.ethereum.summary.EthereumTxDeltaProcessor
 import fund.cyber.address.ethereum.summary.EthereumUncleDeltaProcessor
 import fund.cyber.common.kafka.JsonDeserializer
+import fund.cyber.common.kafka.defaultConsumerConfig
+import fund.cyber.common.with
 import fund.cyber.search.configuration.KAFKA_BROKERS
 import fund.cyber.search.configuration.KAFKA_BROKERS_DEFAULT
 import fund.cyber.search.model.chains.Chain
@@ -125,7 +127,7 @@ class EthereumTxConsumerConfiguration {
         }
     }
 
-    private fun consumerConfigs(): MutableMap<String, Any> = mutableMapOf(
+    private fun consumerConfigs(): MutableMap<String, Any> = defaultConsumerConfig().with(
             ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers,
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
             ConsumerConfig.GROUP_ID_CONFIG to "ethereum-address-summary-update-process",

common-kafka/src/main/kotlin/fund/cyber/common/kafka/Configuration.kt

@@ -0,0 +1,16 @@
+package fund.cyber.common.kafka
+
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.producer.ProducerConfig
+
+const val KAFKA_MAX_MESSAGE_SIZE_BYTES = 15728640
+
+fun defaultConsumerConfig() = mutableMapOf<String,Any>(
+        ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG to KAFKA_MAX_MESSAGE_SIZE_BYTES
+)
+
+fun defaultProducerConfig() = mutableMapOf<String,Any>(
+        ProducerConfig.MAX_REQUEST_SIZE_CONFIG to KAFKA_MAX_MESSAGE_SIZE_BYTES
+)
+
+

common/src/main/kotlin/fund/cyber/common/Collections.kt

@@ -0,0 +1,7 @@
+package fund.cyber.common
+
+
+fun <K, V> MutableMap<K, V>.with(vararg pairs: Pair<K, V>): MutableMap<K, V> {
+    putAll(pairs)
+    return this
+}

dumps/bitcoin/src/main/kotlin/fund/cyber/dump/bitcoin/ApplicationConfiguration.kt

@@ -2,6 +2,8 @@ package fund.cyber.dump.bitcoin
 
 import fund.cyber.cassandra.bitcoin.repository.BitcoinBlockRepository
 import fund.cyber.common.kafka.JsonDeserializer
+import fund.cyber.common.kafka.defaultConsumerConfig
+import fund.cyber.common.with
 import fund.cyber.search.configuration.CHAIN
 import fund.cyber.search.configuration.KAFKA_BROKERS
 import fund.cyber.search.configuration.KAFKA_BROKERS_DEFAULT
@@ -62,7 +64,7 @@ class ApplicationConfiguration {
         return KafkaMessageListenerContainer(consumerFactory, containerProperties)
     }
 
-    private fun consumerConfigs(): MutableMap<String, Any> = mutableMapOf(
+    private fun consumerConfigs(): MutableMap<String, Any> = defaultConsumerConfig().with(
             ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers,
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
             ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to true,

dumps/ethereum/src/main/kotlin/fund/cyber/dump/ethereum/ApplcationConfiguration.kt

@@ -8,6 +8,8 @@ import fund.cyber.cassandra.ethereum.repository.EthereumBlockTxRepository
 import fund.cyber.cassandra.ethereum.repository.EthereumTxRepository
 import fund.cyber.cassandra.ethereum.repository.EthereumUncleRepository
 import fund.cyber.common.kafka.JsonDeserializer
+import fund.cyber.common.kafka.defaultConsumerConfig
+import fund.cyber.common.with
 import fund.cyber.search.configuration.CHAIN
 import fund.cyber.search.configuration.KAFKA_BROKERS
 import fund.cyber.search.configuration.KAFKA_BROKERS_DEFAULT
@@ -138,7 +140,7 @@ class ApplicationConfiguration {
         return KafkaMessageListenerContainer(consumerFactory, containerProperties)
     }
 
-    private fun consumerConfigs(): MutableMap<String, Any> = mutableMapOf(
+    private fun consumerConfigs(): MutableMap<String, Any> = defaultConsumerConfig().with(
             ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers,
             ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
             ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to true,

pumps/bitcoin/src/main/kotlin/fund/cyber/pump/bitcoin/kafka/BitcoinBundleProducerConfiguration.kt

@@ -1,6 +1,8 @@
 package fund.cyber.pump.bitcoin.kafka
 
 import fund.cyber.common.kafka.JsonSerializer
+import fund.cyber.common.kafka.defaultProducerConfig
+import fund.cyber.common.with
 import fund.cyber.search.configuration.KAFKA_BROKERS
 import fund.cyber.search.configuration.KAFKA_BROKERS_DEFAULT
 import fund.cyber.search.model.chains.BitcoinFamilyChain
@@ -67,7 +69,7 @@ class BitcoinBundleProducerConfiguration {
     }
 
     @Bean
-    fun producerConfigs(): Map<String, Any> = mapOf(
+    fun producerConfigs(): Map<String, Any> = defaultProducerConfig().with(
             ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers
     )
 

pumps/ethereum/src/main/kotlin/fund/cyber/pump/ethereum/kafka/EthereumBundleProducerConfiguration.kt

@@ -1,6 +1,8 @@
 package fund.cyber.pump.ethereum.kafka
 
 import fund.cyber.common.kafka.JsonSerializer
+import fund.cyber.common.kafka.defaultProducerConfig
+import fund.cyber.common.with
 import fund.cyber.search.configuration.KAFKA_BROKERS
 import fund.cyber.search.configuration.KAFKA_BROKERS_DEFAULT
 import fund.cyber.search.model.chains.EthereumFamilyChain
@@ -66,7 +68,7 @@ class EthereumBundleProducerConfiguration {
     }
 
     @Bean
-    fun producerConfigs(): Map<String, Any> = mapOf(
+    fun producerConfigs(): Map<String, Any> = defaultProducerConfig().with(
             ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaBrokers
     )