X-Git-Url: https://gerrit.onap.org/r/gitweb?p=dmaap%2Fmessagerouter%2Fmsgrtr.git;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Forg%2Fonap%2Fdmaap%2Fdmf%2Fmr%2Fbackends%2Fkafka%2FKafka011Consumer.java;h=2ec323e00339c278d966c7bd7b82fd85de373388;hp=347f625164fa039bf1d3dc3e99c18e0f337e798a;hb=83746dbc42bad55e52d4bed2617d0d0ca8634cb5;hpb=2b6587421545cf4c21b4b9e211f80c5a394ff697

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 347f625..2ec323e 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
@@ -42,8 +42,7 @@ import org.apache.kafka.common.KafkaException;
 import org.onap.dmaap.dmf.mr.backends.Consumer;
 import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
 
-
-
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
 import com.att.eelf.configuration.EELFLogger;
 import com.att.eelf.configuration.EELFManager;
 
@@ -83,6 +82,12 @@ public class Kafka011Consumer implements Consumer {
 		state = Kafka011Consumer.State.OPENED;
 		kConsumer = cc;
 		fKafkaLiveLockAvoider = klla;
+		
+		String consumerTimeOut = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
+				"consumer.timeout");
+		if (null != consumerTimeOut) {
+			consumerPollTimeOut = Integer.parseInt(consumerTimeOut);
+		}
 		synchronized (kConsumer) {
 			kConsumer.subscribe(Arrays.asList(topic));
 		}
@@ -147,7 +152,7 @@ public class Kafka011Consumer implements Consumer {
 		ExecutorService service = Executors.newSingleThreadExecutor();
 		service.execute(future);
 		try {
-			future.get(5, TimeUnit.SECONDS); // wait 1
+			future.get(consumerPollTimeOut, TimeUnit.SECONDS); // wait 1
 			// second
 		} catch (TimeoutException ex) {
 			// timed out. Try to stop the code if possible.
@@ -370,6 +375,7 @@ public class Kafka011Consumer implements Consumer {
 	private long offset;
 	private Kafka011Consumer.State state;
 	private KafkaLiveLockAvoider2 fKafkaLiveLockAvoider;
+	private int consumerPollTimeOut=5;
 	private static final EELFLogger log = EELFManager.getInstance().getLogger(Kafka011Consumer.class);
 	private final LinkedBlockingQueue<ConsumerRecord<String, String>> fPendingMsgs;