src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/LockInstructionWatcher.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.curator.framework.CuratorFramework;
  27 import org.apache.zookeeper.WatchedEvent;
  28 import org.apache.zookeeper.Watcher;
  29
  30 import java.util.List;
  31
  32 /**
  33  *
  34  * LockInstructionWatcher
  35  * A package-private class used internally by the KafkaLiveLockAvoider.
  36  *
  37  * This class implements the zookeeper Watcher callback and listens for changes on child nodes changing.
  38  * Each child node is actually a Kafka group name that needs to be soft polled.  Deletion of the child nodes
  39  * after soft poll unlocking is finished.
  40  *
  41  *
  42  */
  43 public class LockInstructionWatcher implements Watcher {
  44
  45         private CuratorFramework curatorFramework;
  46         private LiveLockAvoidance avoidanceCallback;
  47         private KafkaLiveLockAvoider2 avoider;
  48
  49         private static final EELFLogger log = EELFManager.getInstance().getLogger(LockInstructionWatcher.class.getName());
  50
  51
  52         public LockInstructionWatcher(CuratorFramework curatorFramework, LiveLockAvoidance avoidanceCallback,
  53                         KafkaLiveLockAvoider2 avoider) {
  54                 super();
  55                 this.curatorFramework = curatorFramework;
  56                 this.avoidanceCallback = avoidanceCallback;
  57                 this.avoider = avoider;
  58         }
  59
  60
  61         @Override
  62         public void process(WatchedEvent event) {
  63
  64                 switch (event.getType()) {
  65                 case NodeChildrenChanged:
  66
  67
  68                         try {
  69
  70                                 log.info("node children changed at path: {}", event.getPath());
  71
  72                                 List<String> children = curatorFramework.getChildren().forPath(event.getPath());
  73
  74                                 log.info("found children nodes prodcessing now");
  75                                 for (String child : children) {
  76                                         String childPath = String.format("%s/%s", event.getPath(), child);
  77                                         log.info("Processing child task at node {}",childPath);
  78                                         avoidanceCallback.handleRebalanceUnlock( child);
  79                                         log.info("Deleting child task at node {}",childPath);
  80                                         curatorFramework.delete().forPath(childPath);
  81                                         }
  82                                 //reset the watch with the avoider
  83                                 avoider.assignNewProcessNode(avoidanceCallback.getAppId(), this);
  84
  85
  86                         } catch (Exception e) {
  87                                 log.error("Error manipulating ZNode data in watcher",e);
  88                         }
  89
  90                         break;
  91
  92                 default:
  93                         log.info("Listner fired on path: {}, with event: {}", event.getPath(), event.getType());
  94                         break;
  95                 }
  96         }
  97
  98
  99 }