diff options
| author |   Brinda Santh <bs2796@att.com> | 2019-10-04 12:23:51 -0400 |
|---|---|---|
| committer | Brinda Santh <bs2796@att.com> | 2019-10-04 12:23:51 -0400 |
| commit | edf18b106492a9199e65d96df98232974446c756 (patch) | |
| tree | 004d7556c978a3fa345113d844154ebee25bb967 /ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin | |
| parent | 4c2b42f141e1501222a284f964eb4303cc4f03aa (diff) | |
Kafka Back pressure configuration
Issue-ID: CCSDK-1666
Signed-off-by: Brinda Santh <bs2796@att.com>
Change-Id: I173672b14e07a89e0bd2d1ea0e47543b37f389f8
Diffstat (limited to 'ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin')
2 files changed, 6 insertions, 0 deletions
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt index ab04054fe..1cd8a2af7 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt @@ -36,6 +36,7 @@ open class KafkaMessageConsumerProperties : MessageConsumerProperties() { var clientId: String? = null var topic: String? = null var pollMillSec: Long = 1000 + var pollRecords: Int = -1 } open class KafkaBasicAuthMessageConsumerProperties : KafkaMessageConsumerProperties() diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaBasicAuthMessageConsumerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaBasicAuthMessageConsumerService.kt index 5a9e61bfd..b5d444a49 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaBasicAuthMessageConsumerService.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaBasicAuthMessageConsumerService.kt @@ -51,6 +51,10 @@ class KafkaBasicAuthMessageConsumerService( if (messageConsumerProperties.clientId != null) { configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = messageConsumerProperties.clientId!! } + /** To handle Back pressure, Get only configured record for processing */ + if (messageConsumerProperties.pollRecords > 0) { + configProperties[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = messageConsumerProperties.pollRecords + } // TODO("Security Implementation based on type") /** add or override already set properties */ additionalConfig?.let { configProperties.putAll(it) } @@ -84,6 +88,7 @@ class KafkaBasicAuthMessageConsumerService( kafkaConsumer!!.use { kc -> while (keepGoing) { val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec)) + log.info("Consumed Records : ${consumerRecords.count()}") runBlocking { consumerRecords?.forEach { consumerRecord -> /** execute the command block */ |