jar/src/main/java/org/onap/music/lockingservice/ProtocolSupport.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   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  *
  11  *     http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *
  19  * ============LICENSE_END=============================================
  20  * ====================================================================
  21  */
  22
  23 package org.onap.music.lockingservice;
  24
  25 import org.apache.zookeeper.CreateMode;
  26 import org.apache.zookeeper.KeeperException;
  27 import org.apache.zookeeper.ZooDefs;
  28 import org.apache.zookeeper.ZooKeeper;
  29 import org.apache.zookeeper.data.ACL;
  30 import org.apache.zookeeper.data.Stat;
  31 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  32 import org.onap.music.eelf.logging.format.AppMessages;
  33 import org.onap.music.eelf.logging.format.ErrorSeverity;
  34 import org.onap.music.eelf.logging.format.ErrorTypes;
  35 import org.onap.music.lockingservice.ZooKeeperOperation;
  36 import java.util.List;
  37 import java.util.concurrent.atomic.AtomicBoolean;
  38
  39 /**
  40  * A base class for protocol implementations which provides a number of higher level helper methods
  41  * for working with ZooKeeper along with retrying synchronous operations if the connection to
  42  * ZooKeeper closes such as {@link #retryOperation(ZooKeeperOperation)}
  43  *
  44  */
  45 class ProtocolSupport {
  46     private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(ProtocolSupport.class);
  47
  48     protected ZooKeeper zookeeper;
  49     private AtomicBoolean closed = new AtomicBoolean(false);
  50     private long retryDelay = 500L;
  51     private int retryCount = 10;
  52     private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
  53
  54     /**
  55      * Closes this strategy and releases any ZooKeeper resources; but keeps the ZooKeeper instance
  56      * open
  57      */
  58     public void close() {
  59         if (closed.compareAndSet(false, true)) {
  60             doClose();
  61         }
  62     }
  63
  64     /**
  65      * return zookeeper client instance
  66      *
  67      * @return zookeeper client instance
  68      */
  69     public ZooKeeper getZookeeper() {
  70         return zookeeper;
  71     }
  72
  73     /**
  74      * return the acl its using
  75      *
  76      * @return the acl.
  77      */
  78     public List<ACL> getAcl() {
  79         return acl;
  80     }
  81
  82     /**
  83      * set the acl
  84      *
  85      * @param acl the acl to set to
  86      */
  87     public void setAcl(List<ACL> acl) {
  88         this.acl = acl;
  89     }
  90
  91     /**
  92      * get the retry delay in milliseconds
  93      *
  94      * @return the retry delay
  95      */
  96     public long getRetryDelay() {
  97         return retryDelay;
  98     }
  99
 100     /**
 101      * Sets the time waited between retry delays
 102      *
 103      * @param retryDelay the retry delay
 104      */
 105     public void setRetryDelay(long retryDelay) {
 106         this.retryDelay = retryDelay;
 107     }
 108
 109     /**
 110      * Allow derived classes to perform some custom closing operations to release resources
 111      */
 112     protected void doClose() {
 113         throw new UnsupportedOperationException();
 114     }
 115
 116
 117     /**
 118      * Perform the given operation, retrying if the connection fails
 119      *
 120      * @return object. it needs to be cast to the callee's expected return type.
 121      * @param operation FILL IN
 122      * @throws KeeperException FILL IN
 123      * @throws InterruptedException FILL IN
 124      */
 125     protected Object retryOperation(ZooKeeperOperation operation)
 126                     throws KeeperException, InterruptedException {
 127         KeeperException exception = null;
 128         for (int i = 0; i < retryCount; i++) {
 129             try {
 130                 return operation.execute();
 131             } catch (KeeperException.SessionExpiredException e) {
 132                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.SESSIONEXPIRED+" for: " + zookeeper + " so reconnecting due to: " + e, ErrorSeverity.ERROR, ErrorTypes.SESSIONEXPIRED);
 133                 throw e;
 134             } catch (KeeperException.ConnectionLossException e) {
 135                 if (exception == null) {
 136                     exception = e;
 137                 }
 138                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.CONNCECTIVITYERROR, ErrorSeverity.ERROR, ErrorTypes.SESSIONEXPIRED);
 139                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),"Attempt " + i + " failed with connection loss so attempting to reconnect: " + e);
 140
 141                 retryDelay(i);
 142             }
 143         }
 144         throw exception;
 145     }
 146
 147     /**
 148      * Ensures that the given path exists with no data, the current ACL and no flags
 149      *
 150      * @param path the lock path
 151      */
 152     protected void ensurePathExists(String path) {
 153         ensureExists(path, null, acl, CreateMode.PERSISTENT);
 154     }
 155
 156     /**
 157      * Ensures that the given path exists with the given data, ACL and flags
 158      *
 159      * @param path the lock path
 160      * @param data the data
 161      * @param acl list of ACLs applying to the path
 162      * @param flags create mode flags
 163      */
 164     protected void ensureExists(final String path, final byte[] data, final List<ACL> acl,
 165                     final CreateMode flags) {
 166         try {
 167             retryOperation(new ZooKeeperOperation() {
 168                 public boolean execute() throws KeeperException, InterruptedException {
 169                     Stat stat = zookeeper.exists(path, false);
 170                     if (stat != null) {
 171                         return true;
 172                     }
 173                     zookeeper.create(path, data, acl, flags);
 174                     return true;
 175                 }
 176             });
 177         } catch (KeeperException e) {
 178             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.KEEPERERROR, ErrorSeverity.ERROR, ErrorTypes.LOCKINGERROR);
 179         } catch (InterruptedException e) {
 180             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.EXECUTIONINTERRUPTED, ErrorSeverity.ERROR, ErrorTypes.LOCKINGERROR);
 181         }
 182     }
 183
 184     /**
 185      * Returns true if this protocol has been closed
 186      *
 187      * @return true if this protocol is closed
 188      */
 189     protected boolean isClosed() {
 190         return closed.get();
 191     }
 192
 193     /**
 194      * Performs a retry delay if this is not the first attempt
 195      *
 196      * @param attemptCount the number of the attempts performed so far
 197      */
 198     protected void retryDelay(int attemptCount) {
 199         if (attemptCount > 0) {
 200             try {
 201                 Thread.sleep(attemptCount * retryDelay);
 202             } catch (InterruptedException e) {
 203                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.EXECUTIONINTERRUPTED, ErrorSeverity.ERROR, ErrorTypes.GENERALSERVICEERROR);
 204                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),"Thread failed to sleep: " + e);
 205                 Thread.currentThread().interrupt();
 206             }
 207         }
 208     }
 209 }