src/main/java/com/att/dmf/mr/backends/kafka/Kafka011ConsumerUtil.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *        http://www.apache.org/licenses/LICENSE-2.0
  11 *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package com.att.dmf.mr.backends.kafka;
  23
  24 import java.util.ArrayList;
  25
  26 import org.apache.kafka.clients.consumer.ConsumerRecords;
  27
  28 //import org.slf4j.Logger;
  29 //import org.slf4j.LoggerFactory;
  30
  31 import com.att.eelf.configuration.EELFLogger;
  32 import com.att.eelf.configuration.EELFManager;
  33
  34 /**
  35  * A consumer Util class for force polling when a rebalance issue is anticipated
  36  *
  37  * @author Ram
  38  *
  39  */
  40 public class Kafka011ConsumerUtil {
  41         private static final EELFLogger log = EELFManager.getInstance().getLogger(Kafka011ConsumerUtil.class);
  42
  43         /**
  44          * @param fconsumercache
  45          * @param fTopic
  46          * @param fGroup
  47          * @param fId
  48          * @return
  49          */
  50         public static boolean forcePollOnConsumer(final String fTopic, final String fGroup, final String fId) {
  51
  52                 Thread forcepollThread = new Thread(new Runnable() {
  53                         public void run() {
  54                                 try {
  55
  56                                         ArrayList<Kafka011Consumer> kcsList = null;
  57
  58                                         kcsList = KafkaConsumerCache.getInstance().getConsumerListForCG(fTopic + "::" + fGroup + "::", fId);
  59                                         if (null != kcsList) {
  60                                                 for (int counter = 0; counter < kcsList.size(); counter++) {
  61
  62                                                         Kafka011Consumer kc1 = kcsList.get(counter);
  63
  64                                                         try {
  65                                                                 ConsumerRecords<String, String> recs = kc1.getConsumer().poll(0);
  66                                                                 log.info("soft poll on " + kc1);
  67                                                         } catch (java.util.ConcurrentModificationException e) {
  68                                                                 log.error("Error occurs for " + e);
  69                                                         }
  70
  71                                                 }
  72
  73                                         }
  74
  75                                 } catch (Exception e) {
  76                                         log.error("Failed and go to Exception block for " + fGroup + " " + e.getMessage());
  77                                         e.printStackTrace();
  78                                 }
  79                         }
  80                 });
  81
  82                 forcepollThread.start();
  83
  84                 return false;
  85
  86         }
  87
  88         /**
  89          * @param fconsumercache
  90          * @param group
  91          * @return
  92          */
  93         public static boolean forcePollOnConsumer(final String group) {
  94
  95                 Thread forcepollThread = new Thread(new Runnable() {
  96                         public void run() {
  97                                 try {
  98                                         ArrayList<Kafka011Consumer> kcsList = new ArrayList<Kafka011Consumer>();
  99                                         kcsList = KafkaConsumerCache.getInstance().getConsumerListForCG(group);
 100
 101                                         if (null != kcsList) {
 102
 103                                                 for (int counter = 0; counter < kcsList.size(); counter++) {
 104
 105                                                         Kafka011Consumer kc1 = kcsList.get(counter);
 106                                                         log.info("soft poll on remote nodes " + kc1);
 107                                                         ConsumerRecords<String, String> recs = kc1.getConsumer().poll(0);
 108                                                 }
 109
 110                                         }
 111
 112                                 } catch (java.util.ConcurrentModificationException e) {
 113                                         log.error("Error occurs for " + e);
 114                                 } catch (Exception e) {
 115                                         log.error("Failed and go to Exception block for " + group + " " + e.getMessage());
 116                                         e.printStackTrace();
 117                                 }
 118                         }
 119                 });
 120
 121                 forcepollThread.start();
 122                 return false;
 123
 124         }
 125
 126 }