src/main/java/org/onap/dmaap/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 org.onap.dmaap.dmf.mr.backends.kafka;
  23
  24 import java.util.ArrayList;
  25
  26 import org.apache.kafka.clients.consumer.ConsumerRecords;
  27
  28
  29
  30 import com.att.eelf.configuration.EELFLogger;
  31 import com.att.eelf.configuration.EELFManager;
  32
  33 /**
  34  * A consumer Util class for force polling when a rebalance issue is anticipated
  35  *
  36  * @author Ram
  37  *
  38  */
  39 public class Kafka011ConsumerUtil {
  40         private static final EELFLogger log = EELFManager.getInstance().getLogger(Kafka011ConsumerUtil.class);
  41
  42         /**
  43          * @param fconsumercache
  44          * @param fTopic
  45          * @param fGroup
  46          * @param fId
  47          * @return
  48          */
  49         public static boolean forcePollOnConsumer(final String fTopic, final String fGroup, final String fId) {
  50
  51                 Thread forcepollThread = new Thread(new Runnable() {
  52                         public void run() {
  53                                 try {
  54
  55                                         ArrayList<Kafka011Consumer> kcsList = null;
  56
  57                                         kcsList = KafkaConsumerCache.getInstance().getConsumerListForCG(fTopic + "::" + fGroup + "::", fId);
  58                                         if (null != kcsList) {
  59                                                 for (int counter = 0; counter < kcsList.size(); counter++) {
  60
  61                                                         Kafka011Consumer kc1 = kcsList.get(counter);
  62
  63                                                         try {
  64                                                                 ConsumerRecords<String, String> recs = kc1.getConsumer().poll(0);
  65                                                                 log.info("soft poll on " + kc1);
  66                                                         } catch (java.util.ConcurrentModificationException e) {
  67                                                                 log.error("Error occurs for " + e);
  68                                                         }
  69
  70                                                 }
  71
  72                                         }
  73
  74                                 } catch (Exception e) {
  75                                         log.error("Failed and go to Exception block for " + fGroup + " " + e.getMessage());
  76                                 }
  77                         }
  78                 });
  79
  80                 forcepollThread.start();
  81
  82                 return false;
  83
  84         }
  85
  86         /**
  87          * @param fconsumercache
  88          * @param group
  89          * @return
  90          */
  91         public static boolean forcePollOnConsumer(final String group) {
  92
  93                 Thread forcepollThread = new Thread(new Runnable() {
  94                         public void run() {
  95                                 try {
  96                                         ArrayList<Kafka011Consumer> kcsList = new ArrayList<Kafka011Consumer>();
  97                                         kcsList = KafkaConsumerCache.getInstance().getConsumerListForCG(group);
  98
  99                                         if (null != kcsList) {
 100
 101                                                 for (int counter = 0; counter < kcsList.size(); counter++) {
 102
 103                                                         Kafka011Consumer kc1 = kcsList.get(counter);
 104                                                         log.info("soft poll on remote nodes " + kc1);
 105                                                         ConsumerRecords<String, String> recs = kc1.getConsumer().poll(0);
 106                                                 }
 107
 108                                         }
 109
 110                                 } catch (java.util.ConcurrentModificationException e) {
 111                                         log.error("Error occurs for " + e);
 112                                 } catch (Exception e) {
 113                                         log.error("Failed and go to Exception block for " + group + " " + e.getMessage());
 114                                 }
 115                         }
 116                 });
 117
 118                 forcepollThread.start();
 119                 return false;
 120
 121         }
 122
 123 }