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