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
  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                                         e.printStackTrace();
  77                                 }
  78                         }
  79                 });
  80
  81                 forcepollThread.start();
  82
  83                 return false;
  84
  85         }
  86
  87         /**
  88          * @param fconsumercache
  89          * @param group
  90          * @return
  91          */
  92         public static boolean forcePollOnConsumer(final String group) {
  93
  94                 Thread forcepollThread = new Thread(new Runnable() {
  95                         public void run() {
  96                                 try {
  97                                         ArrayList<Kafka011Consumer> kcsList = new ArrayList<Kafka011Consumer>();
  98                                         kcsList = KafkaConsumerCache.getInstance().getConsumerListForCG(group);
  99
 100                                         if (null != kcsList) {
 101
 102                                                 for (int counter = 0; counter < kcsList.size(); counter++) {
 103
 104                                                         Kafka011Consumer kc1 = kcsList.get(counter);
 105                                                         log.info("soft poll on remote nodes " + kc1);
 106                                                         ConsumerRecords<String, String> recs = kc1.getConsumer().poll(0);
 107                                                 }
 108
 109                                         }
 110
 111                                 } catch (java.util.ConcurrentModificationException e) {
 112                                         log.error("Error occurs for " + e);
 113                                 } catch (Exception e) {
 114                                         log.error("Failed and go to Exception block for " + group + " " + e.getMessage());
 115                                         e.printStackTrace();
 116                                 }
 117                         }
 118                 });
 119
 120                 forcepollThread.start();
 121                 return false;
 122
 123         }
 124
 125 }