import org.apache.kafka.clients.producer.ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION
import org.apache.kafka.clients.producer.ProducerConfig.RETRIES_CONFIG
import org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG
-import org.onap.dcae.collectors.veshv.boundary.Sink
+import org.apache.kafka.clients.producer.ProducerConfig.MAX_REQUEST_SIZE_CONFIG
+import org.apache.kafka.clients.producer.ProducerConfig.BUFFER_MEMORY_CONFIG
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.KafkaConfiguration
import org.onap.ves.VesEventOuterClass.CommonEventHeader
import reactor.kafka.sender.KafkaSender
import reactor.kafka.sender.SenderOptions
+import java.lang.Integer.max
/**
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
override fun invoke(ctx: ClientContext) = KafkaSink(kafkaSender, ctx)
companion object {
+ private const val MAXIMUM_REQUEST_SIZE_MULTIPLIER = 1.2f
+ private const val BUFFER_MEMORY_MULTIPLIER = 32
+ private const val MINIMUM_BUFFER_MEMORY = 32 * 1024 * 1024
private fun constructKafkaSender(config: KafkaConfiguration) =
KafkaSender.create(constructSenderOptions(config))
private fun constructSenderOptions(config: KafkaConfiguration) =
SenderOptions.create<CommonEventHeader, VesMessage>()
.producerProperty(BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers)
+ .producerProperty(MAX_REQUEST_SIZE_CONFIG, maxRequestSize(config))
+ .producerProperty(BUFFER_MEMORY_CONFIG, bufferMemory(config))
.producerProperty(KEY_SERIALIZER_CLASS_CONFIG, ProtobufSerializer::class.java)
.producerProperty(VALUE_SERIALIZER_CLASS_CONFIG, VesMessageSerializer::class.java)
.producerProperty(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1)
.producerProperty(RETRIES_CONFIG, 1)
.producerProperty(ACKS_CONFIG, "1")
.stopOnError(false)
+
+ private fun maxRequestSize(config: KafkaConfiguration) =
+ (MAXIMUM_REQUEST_SIZE_MULTIPLIER * config.maximalRequestSizeBytes).toInt()
+
+ private fun bufferMemory(config: KafkaConfiguration) =
+ max(MINIMUM_BUFFER_MEMORY, BUFFER_MEMORY_MULTIPLIER * config.maximalRequestSizeBytes)
}
}
val configurationProviderParams = createConfigurationProviderParams(cmdLine).bind()
ServerConfiguration(
serverListenAddress = InetSocketAddress(listenPort),
- kafkaConfiguration = KafkaConfiguration(kafkaServers),
+ kafkaConfiguration = KafkaConfiguration(kafkaServers, maxPayloadSizeBytes),
healthCheckApiListenAddress = InetSocketAddress(healthCheckApiPort),
configurationProviderParams = configurationProviderParams,
securityConfiguration = security,