From e157241cb7dfe023f57532f829e381644bdf18be Mon Sep 17 00:00:00 2001 From: sunil unnava Date: Tue, 11 Dec 2018 16:25:26 -0500 Subject: add configurable default partitions and replicas Issue-ID: DMAAP-903 Change-Id: Iabb3da85c3e42ddf68d049e6a8164449f7be8296 Signed-off-by: sunil unnava --- .../dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java | 3 ++- .../onap/dmaap/dmf/mr/service/impl/TopicServiceImpl.java | 15 +++++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) (limited to 'src/main') diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java index 2ec323e..93374fb 100644 --- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java +++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java @@ -34,6 +34,7 @@ import java.util.concurrent.RunnableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import org.apache.commons.lang.StringUtils; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; @@ -85,7 +86,7 @@ public class Kafka011Consumer implements Consumer { String consumerTimeOut = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "consumer.timeout"); - if (null != consumerTimeOut) { + if (StringUtils.isNotEmpty(consumerTimeOut)) { consumerPollTimeOut = Integer.parseInt(consumerTimeOut); } synchronized (kConsumer) { diff --git a/src/main/java/org/onap/dmaap/dmf/mr/service/impl/TopicServiceImpl.java b/src/main/java/org/onap/dmaap/dmf/mr/service/impl/TopicServiceImpl.java index f2ba222..626828b 100644 --- a/src/main/java/org/onap/dmaap/dmf/mr/service/impl/TopicServiceImpl.java +++ b/src/main/java/org/onap/dmaap/dmf/mr/service/impl/TopicServiceImpl.java @@ -26,6 +26,7 @@ package org.onap.dmaap.dmf.mr.service.impl; import java.io.IOException; +import org.apache.commons.lang.StringUtils; import org.apache.http.HttpStatus; import org.json.JSONArray; import org.json.JSONException; @@ -281,14 +282,28 @@ public class TopicServiceImpl implements TopicService { final String desc = topicBean.getTopicDescription(); int partition = topicBean.getPartitionCount(); // int replica = topicBean.getReplicationCount(); + String defaultPartitions = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, + "default.partitions"); + String defaultReplicas = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, + "default.replicas"); if (partition == 0) { + if(StringUtils.isNotEmpty(defaultPartitions)){ + partition=Integer.parseInt(defaultPartitions); + } + else{ partition = 1; + } } final int partitions = partition; int replica = topicBean.getReplicationCount(); if (replica == 0) { + if(StringUtils.isNotEmpty(defaultReplicas)){ + replica=Integer.parseInt(defaultReplicas); + } + else{ replica = 1; + } } final int replicas = replica; boolean transactionEnabled = topicBean.isTransactionEnabled(); -- cgit 1.2.3-korg