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