feature-distributed-locking/src/main/java/org/onap/policy/distributed/locking/DistributedLockManager.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2019 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  *
  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  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.distributed.locking;
  22
  23 import java.sql.Connection;
  24 import java.sql.PreparedStatement;
  25 import java.sql.ResultSet;
  26 import java.sql.SQLException;
  27 import java.sql.SQLTransientException;
  28 import java.util.HashSet;
  29 import java.util.Map;
  30 import java.util.Properties;
  31 import java.util.Set;
  32 import java.util.UUID;
  33 import java.util.concurrent.RejectedExecutionException;
  34 import java.util.concurrent.ScheduledExecutorService;
  35 import java.util.concurrent.ScheduledFuture;
  36 import java.util.concurrent.TimeUnit;
  37 import java.util.concurrent.atomic.AtomicBoolean;
  38 import java.util.concurrent.atomic.AtomicReference;
  39 import lombok.AccessLevel;
  40 import lombok.Getter;
  41 import lombok.Setter;
  42 import org.apache.commons.dbcp2.BasicDataSource;
  43 import org.apache.commons.dbcp2.BasicDataSourceFactory;
  44 import org.onap.policy.common.utils.network.NetworkUtil;
  45 import org.onap.policy.drools.core.lock.LockCallback;
  46 import org.onap.policy.drools.core.lock.LockState;
  47 import org.onap.policy.drools.core.lock.PolicyResourceLockManager;
  48 import org.onap.policy.drools.features.PolicyEngineFeatureApi;
  49 import org.onap.policy.drools.persistence.SystemPersistenceConstants;
  50 import org.onap.policy.drools.system.PolicyEngine;
  51 import org.onap.policy.drools.system.PolicyEngineConstants;
  52 import org.onap.policy.drools.system.internal.FeatureLockImpl;
  53 import org.onap.policy.drools.system.internal.LockManager;
  54 import org.slf4j.Logger;
  55 import org.slf4j.LoggerFactory;
  56
  57
  58 /**
  59  * Distributed implementation of the Lock Feature. Maintains locks across servers using a
  60  * shared DB.
  61  *
  62  * <p/>
  63  * Note: this implementation does <i>not</i> honor the waitForLocks={@code true}
  64  * parameter.
  65  *
  66  * <p/>
  67  * Additional Notes:
  68  * <dl>
  69  * <li>The <i>owner</i> field in the DB is not derived from the lock's owner info, but is
  70  * instead populated with the {@link #uuidString}.</li>
  71  * <li>A periodic check of the DB is made to determine if any of the locks have
  72  * expired.</li>
  73  * <li>When a lock is deserialized, it will not initially appear in this feature's map; it
  74  * will be added to the map once free() or extend() is invoked, provided there isn't
  75  * already an entry. In addition, it initially has the host and UUID of the feature
  76  * instance that created it. However, as soon as doExtend() completes successfully, the
  77  * host and UUID of the lock will be updated to reflect the values within this feature
  78  * instance.</li>
  79  * </dl>
  80  */
  81 public class DistributedLockManager extends LockManager<DistributedLockManager.DistributedLock>
  82                 implements PolicyEngineFeatureApi {
  83
  84     private static final Logger logger = LoggerFactory.getLogger(DistributedLockManager.class);
  85
  86     private static final String CONFIGURATION_PROPERTIES_NAME = "feature-distributed-locking";
  87
  88     @Getter(AccessLevel.PROTECTED)
  89     @Setter(AccessLevel.PROTECTED)
  90     private static DistributedLockManager latestInstance = null;
  91
  92
  93     /**
  94      * Name of the host on which this JVM is running.
  95      */
  96     @Getter
  97     private final String hostName;
  98
  99     /**
 100      * UUID of this object.
 101      */
 102     @Getter
 103     private final String uuidString = UUID.randomUUID().toString();
 104
 105     /**
 106      * Maps a resource to the lock that owns it, or is awaiting a request for it. Once a
 107      * lock is added to the map, it remains in the map until the lock is lost or until the
 108      * unlock request completes.
 109      */
 110     private final Map<String, DistributedLock> resource2lock;
 111
 112     /**
 113      * Thread pool used to check for lock expiration and to notify owners when locks are
 114      * granted or lost.
 115      */
 116     private ScheduledExecutorService exsvc = null;
 117
 118     /**
 119      * Used to cancel the expiration checker on shutdown.
 120      */
 121     private ScheduledFuture<?> checker = null;
 122
 123     /**
 124      * Feature properties.
 125      */
 126     private DistributedLockProperties featProps;
 127
 128     /**
 129      * Data source used to connect to the DB.
 130      */
 131     private BasicDataSource dataSource = null;
 132
 133
 134     /**
 135      * Constructs the object.
 136      */
 137     public DistributedLockManager() {
 138         this.hostName = NetworkUtil.getHostname();
 139         this.resource2lock = getResource2lock();
 140     }
 141
 142     @Override
 143     public int getSequenceNumber() {
 144         return 1000;
 145     }
 146
 147     @Override
 148     public PolicyResourceLockManager beforeCreateLockManager(PolicyEngine engine, Properties properties) {
 149
 150         try {
 151             this.featProps = new DistributedLockProperties(getProperties(CONFIGURATION_PROPERTIES_NAME));
 152             this.dataSource = makeDataSource();
 153
 154             return this;
 155
 156         } catch (Exception e) {
 157             throw new DistributedLockManagerException(e);
 158         }
 159     }
 160
 161     @Override
 162     public boolean afterStart(PolicyEngine engine) {
 163
 164         try {
 165             exsvc = PolicyEngineConstants.getManager().getExecutorService();
 166             exsvc.execute(this::deleteExpiredDbLocks);
 167             checker = exsvc.schedule(this::checkExpired, featProps.getExpireCheckSec(), TimeUnit.SECONDS);
 168
 169             setLatestInstance(this);
 170
 171         } catch (Exception e) {
 172             throw new DistributedLockManagerException(e);
 173         }
 174
 175         return false;
 176     }
 177
 178     /**
 179      * Make data source.
 180      *
 181      * @return a new, pooled data source
 182      * @throws Exception exception
 183      */
 184     protected BasicDataSource makeDataSource() throws Exception {
 185         Properties props = new Properties();
 186         props.put("driverClassName", featProps.getDbDriver());
 187         props.put("url", featProps.getDbUrl());
 188         props.put("username", featProps.getDbUser());
 189         props.put("password", featProps.getDbPwd());
 190         props.put("testOnBorrow", "true");
 191         props.put("poolPreparedStatements", "true");
 192
 193         // additional properties are listed in the GenericObjectPool API
 194
 195         return BasicDataSourceFactory.createDataSource(props);
 196     }
 197
 198     /**
 199      * Deletes expired locks from the DB.
 200      */
 201     private void deleteExpiredDbLocks() {
 202         logger.info("deleting all expired locks from the DB");
 203
 204         try (Connection conn = dataSource.getConnection();
 205                         PreparedStatement stmt = conn
 206                                         .prepareStatement("DELETE FROM pooling.locks WHERE expirationTime <= now()")) {
 207
 208             int ndel = stmt.executeUpdate();
 209             logger.info("deleted {} expired locks from the DB", ndel);
 210
 211         } catch (SQLException e) {
 212             logger.warn("failed to delete expired locks from the DB", e);
 213         }
 214     }
 215
 216     /**
 217      * Closes the data source. Does <i>not</i> invoke any lock call-backs.
 218      */
 219     @Override
 220     public boolean afterStop(PolicyEngine engine) {
 221         exsvc = null;
 222
 223         if (checker != null) {
 224             checker.cancel(true);
 225         }
 226
 227         closeDataSource();
 228         return false;
 229     }
 230
 231     /**
 232      * Closes {@link #dataSource} and sets it to {@code null}.
 233      */
 234     private void closeDataSource() {
 235         try {
 236             if (dataSource != null) {
 237                 dataSource.close();
 238             }
 239
 240         } catch (SQLException e) {
 241             logger.error("cannot close the distributed locking DB", e);
 242         }
 243
 244         dataSource = null;
 245     }
 246
 247     @Override
 248     protected boolean hasInstanceChanged() {
 249         return (getLatestInstance() != this);
 250     }
 251
 252     @Override
 253     protected void finishLock(DistributedLock lock) {
 254         lock.scheduleRequest(lock::doLock);
 255     }
 256
 257     /**
 258      * Checks for expired locks.
 259      */
 260     private void checkExpired() {
 261         try {
 262             logger.info("checking for expired locks");
 263             Set<String> expiredIds = new HashSet<>(resource2lock.keySet());
 264             identifyDbLocks(expiredIds);
 265             expireLocks(expiredIds);
 266
 267             checker = exsvc.schedule(this::checkExpired, featProps.getExpireCheckSec(), TimeUnit.SECONDS);
 268
 269         } catch (RejectedExecutionException e) {
 270             logger.warn("thread pool is no longer accepting requests", e);
 271
 272         } catch (SQLException | RuntimeException e) {
 273             logger.error("error checking expired locks", e);
 274
 275             if (isAlive()) {
 276                 checker = exsvc.schedule(this::checkExpired, featProps.getRetrySec(), TimeUnit.SECONDS);
 277             }
 278         }
 279
 280         logger.info("done checking for expired locks");
 281     }
 282
 283     /**
 284      * Identifies this feature instance's locks that the DB indicates are still active.
 285      *
 286      * @param expiredIds IDs of resources that have expired locks. If a resource is still
 287      *        locked, it's ID is removed from this set
 288      * @throws SQLException if a DB error occurs
 289      */
 290     private void identifyDbLocks(Set<String> expiredIds) throws SQLException {
 291         /*
 292          * We could query for host and UUIDs that actually appear within the locks, but
 293          * those might change while the query is running so no real value in doing that.
 294          * On the other hand, there's only a brief instance between the time a
 295          * deserialized lock is added to this feature instance and its doExtend() method
 296          * updates its host and UUID to match this feature instance. If this happens to
 297          * run during that brief instance, then the lock will be lost and the callback
 298          * invoked. It isn't worth complicating this code further to handle those highly
 299          * unlikely cases.
 300          */
 301
 302         // @formatter:off
 303         try (Connection conn = dataSource.getConnection();
 304                     PreparedStatement stmt = conn.prepareStatement(
 305                         "SELECT resourceId FROM pooling.locks WHERE host=? AND owner=? AND expirationTime > now()")) {
 306             // @formatter:on
 307
 308             stmt.setString(1, hostName);
 309             stmt.setString(2, uuidString);
 310
 311             try (ResultSet resultSet = stmt.executeQuery()) {
 312                 while (resultSet.next()) {
 313                     String resourceId = resultSet.getString(1);
 314
 315                     // we have now seen this resource id
 316                     expiredIds.remove(resourceId);
 317                 }
 318             }
 319         }
 320     }
 321
 322     /**
 323      * Expires locks for the resources that no longer appear within the DB.
 324      *
 325      * @param expiredIds IDs of resources that have expired locks
 326      */
 327     private void expireLocks(Set<String> expiredIds) {
 328         for (String resourceId : expiredIds) {
 329             AtomicReference<DistributedLock> lockref = new AtomicReference<>(null);
 330
 331             resource2lock.computeIfPresent(resourceId, (key, lock) -> {
 332                 if (lock.isActive()) {
 333                     // it thinks it's active, but it isn't - remove from the map
 334                     lockref.set(lock);
 335                     return null;
 336                 }
 337
 338                 return lock;
 339             });
 340
 341             DistributedLock lock = lockref.get();
 342             if (lock != null) {
 343                 logger.debug("removed lock from map {}", lock);
 344                 lock.deny(DistributedLock.LOCK_LOST_MSG, false);
 345             }
 346         }
 347     }
 348
 349     /**
 350      * Distributed Lock implementation.
 351      */
 352     public static class DistributedLock extends FeatureLockImpl {
 353         private static final String SQL_FAILED_MSG = "request failed for lock: {}";
 354
 355         private static final long serialVersionUID = 1L;
 356
 357         /**
 358          * Feature containing this lock. May be {@code null} until the feature is
 359          * identified. Note: this can only be null if the lock has been de-serialized.
 360          */
 361         private transient DistributedLockManager feature;
 362
 363         /**
 364          * Host name from the feature instance that created this object. Replaced with the
 365          * host name from the current feature instance whenever the lock is successfully
 366          * extended.
 367          */
 368         private String hostName;
 369
 370         /**
 371          * UUID string from the feature instance that created this object. Replaced with
 372          * the UUID string from the current feature instance whenever the lock is
 373          * successfully extended.
 374          */
 375         private String uuidString;
 376
 377         /**
 378          * {@code True} if the lock is busy making a request, {@code false} otherwise.
 379          */
 380         private transient boolean busy = false;
 381
 382         /**
 383          * Request to be performed.
 384          */
 385         private transient RunnableWithEx request = null;
 386
 387         /**
 388          * Number of times we've retried a request.
 389          */
 390         private transient int nretries = 0;
 391
 392         /**
 393          * Constructs the object.
 394          */
 395         public DistributedLock() {
 396             this.feature = null;
 397             this.hostName = "";
 398             this.uuidString = "";
 399         }
 400
 401         /**
 402          * Constructs the object.
 403          *
 404          * @param state initial state of the lock
 405          * @param resourceId identifier of the resource to be locked
 406          * @param ownerKey information identifying the owner requesting the lock
 407          * @param holdSec amount of time, in seconds, for which the lock should be held,
 408          *        after which it will automatically be released
 409          * @param callback callback to be invoked once the lock is granted, or
 410          *        subsequently lost; must not be {@code null}
 411          * @param feature feature containing this lock
 412          */
 413         public DistributedLock(LockState state, String resourceId, String ownerKey, int holdSec, LockCallback callback,
 414                         DistributedLockManager feature) {
 415             super(state, resourceId, ownerKey, holdSec, callback);
 416
 417             this.feature = feature;
 418             this.hostName = feature.hostName;
 419             this.uuidString = feature.uuidString;
 420         }
 421
 422         @Override
 423         public boolean free() {
 424             if (!freeAllowed()) {
 425                 return false;
 426             }
 427
 428             AtomicBoolean result = new AtomicBoolean(false);
 429
 430             feature.resource2lock.computeIfPresent(getResourceId(), (resourceId, curlock) -> {
 431                 if (curlock == this && !isUnavailable()) {
 432                     // this lock was the owner
 433                     result.set(true);
 434                     setState(LockState.UNAVAILABLE);
 435
 436                     /*
 437                      * NOTE: do NOT return null; curlock must remain until doUnlock
 438                      * completes.
 439                      */
 440                 }
 441
 442                 return curlock;
 443             });
 444
 445             if (result.get()) {
 446                 scheduleRequest(this::doUnlock);
 447                 return true;
 448             }
 449
 450             return false;
 451         }
 452
 453         @Override
 454         public void extend(int holdSec, LockCallback callback) {
 455             if (!extendAllowed(holdSec, callback)) {
 456                 return;
 457             }
 458
 459             AtomicBoolean success = new AtomicBoolean(false);
 460
 461             feature.resource2lock.computeIfPresent(getResourceId(), (resourceId, curlock) -> {
 462                 if (curlock == this && !isUnavailable()) {
 463                     success.set(true);
 464                     setState(LockState.WAITING);
 465                 }
 466
 467                 // note: leave it in the map until doUnlock() removes it
 468
 469                 return curlock;
 470             });
 471
 472             if (success.get()) {
 473                 scheduleRequest(this::doExtend);
 474
 475             } else {
 476                 deny(NOT_LOCKED_MSG, true);
 477             }
 478         }
 479
 480         @Override
 481         protected boolean addToFeature() {
 482             feature = getLatestInstance();
 483             if (feature == null) {
 484                 logger.warn("no feature yet for {}", this);
 485                 return false;
 486             }
 487
 488             // put this lock into the map
 489             feature.resource2lock.putIfAbsent(getResourceId(), this);
 490
 491             return true;
 492         }
 493
 494         /**
 495          * Schedules a request for execution.
 496          *
 497          * @param schedreq the request that should be scheduled
 498          */
 499         private synchronized void scheduleRequest(RunnableWithEx schedreq) {
 500             logger.debug("schedule lock action {}", this);
 501             nretries = 0;
 502             request = schedreq;
 503             getThreadPool().execute(this::doRequest);
 504         }
 505
 506         /**
 507          * Reschedules a request for execution, if there is not already a request in the
 508          * queue, and if the retry count has not been exhausted.
 509          *
 510          * @param req request to be rescheduled
 511          */
 512         private void rescheduleRequest(RunnableWithEx req) {
 513             synchronized (this) {
 514                 if (request != null) {
 515                     // a new request has already been scheduled - it supersedes "req"
 516                     logger.debug("not rescheduling lock action {}", this);
 517                     return;
 518                 }
 519
 520                 if (nretries++ < feature.featProps.getMaxRetries()) {
 521                     logger.debug("reschedule for {}s {}", feature.featProps.getRetrySec(), this);
 522                     request = req;
 523                     getThreadPool().schedule(this::doRequest, feature.featProps.getRetrySec(), TimeUnit.SECONDS);
 524                     return;
 525                 }
 526             }
 527
 528             logger.warn("retry count {} exhausted for lock: {}", feature.featProps.getMaxRetries(), this);
 529             removeFromMap();
 530         }
 531
 532         /**
 533          * Gets, and removes, the next request from the queue. Clears {@link #busy} if
 534          * there are no more requests in the queue.
 535          *
 536          * @param prevReq the previous request that was just run
 537          *
 538          * @return the next request, or {@code null} if the queue is empty
 539          */
 540         private synchronized RunnableWithEx getNextRequest(RunnableWithEx prevReq) {
 541             if (request == null || request == prevReq) {
 542                 logger.debug("no more requests for {}", this);
 543                 busy = false;
 544                 return null;
 545             }
 546
 547             RunnableWithEx req = request;
 548             request = null;
 549
 550             return req;
 551         }
 552
 553         /**
 554          * Executes the current request, if none are currently executing.
 555          */
 556         private void doRequest() {
 557             synchronized (this) {
 558                 if (busy) {
 559                     // another thread is already processing the request(s)
 560                     return;
 561                 }
 562                 busy = true;
 563             }
 564
 565             /*
 566              * There is a race condition wherein this thread could invoke run() while the
 567              * next scheduled thread checks the busy flag and finds that work is being
 568              * done and returns, leaving the next work item in "request". In that case,
 569              * the next work item may never be executed, thus we use a loop here, instead
 570              * of just executing a single request.
 571              */
 572             RunnableWithEx req = null;
 573             while ((req = getNextRequest(req)) != null) {
 574                 if (feature.resource2lock.get(getResourceId()) != this) {
 575                     /*
 576                      * no longer in the map - don't apply the action, as it may interfere
 577                      * with any newly added Lock object
 578                      */
 579                     logger.debug("discard lock action {}", this);
 580                     synchronized (this) {
 581                         busy = false;
 582                     }
 583                     return;
 584                 }
 585
 586                 try {
 587                     /*
 588                      * Run the request. If it throws an exception, then it will be
 589                      * rescheduled for execution a little later.
 590                      */
 591                     req.run();
 592
 593                 } catch (SQLException e) {
 594                     logger.warn(SQL_FAILED_MSG, this, e);
 595
 596                     if (e.getCause() instanceof SQLTransientException) {
 597                         // retry the request a little later
 598                         rescheduleRequest(req);
 599                     } else {
 600                         removeFromMap();
 601                     }
 602
 603                 } catch (RuntimeException e) {
 604                     logger.warn(SQL_FAILED_MSG, this, e);
 605                     removeFromMap();
 606                 }
 607             }
 608         }
 609
 610         /**
 611          * Attempts to add a lock to the DB. Generates a callback, indicating success or
 612          * failure.
 613          *
 614          * @throws SQLException if a DB error occurs
 615          */
 616         private void doLock() throws SQLException {
 617             if (!isWaiting()) {
 618                 logger.debug("discard doLock {}", this);
 619                 return;
 620             }
 621
 622             /*
 623              * There is a small window in which a client could invoke free() before the DB
 624              * is updated. In that case, doUnlock will be added to the queue to run after
 625              * this, which will delete the record, as desired. In addition, grant() will
 626              * not do anything, because the lock state will have been set to UNAVAILABLE
 627              * by free().
 628              */
 629
 630             logger.debug("doLock {}", this);
 631             try (Connection conn = feature.dataSource.getConnection()) {
 632                 boolean success = false;
 633                 try {
 634                     success = doDbInsert(conn);
 635
 636                 } catch (SQLException e) {
 637                     logger.info("failed to insert lock record - attempting update: {}", this, e);
 638                     success = doDbUpdate(conn);
 639                 }
 640
 641                 if (success) {
 642                     grant(true);
 643                     return;
 644                 }
 645             }
 646
 647             removeFromMap();
 648         }
 649
 650         /**
 651          * Attempts to remove a lock from the DB. Does <i>not</i> generate a callback if
 652          * it fails, as this should only be executed in response to a call to
 653          * {@link #free()}.
 654          *
 655          * @throws SQLException if a DB error occurs
 656          */
 657         private void doUnlock() throws SQLException {
 658             logger.debug("unlock {}", this);
 659             try (Connection conn = feature.dataSource.getConnection()) {
 660                 doDbDelete(conn);
 661             }
 662
 663             removeFromMap();
 664         }
 665
 666         /**
 667          * Attempts to extend a lock in the DB. Generates a callback, indicating success
 668          * or failure.
 669          *
 670          * @throws SQLException if a DB error occurs
 671          */
 672         private void doExtend() throws SQLException {
 673             if (!isWaiting()) {
 674                 logger.debug("discard doExtend {}", this);
 675                 return;
 676             }
 677
 678             /*
 679              * There is a small window in which a client could invoke free() before the DB
 680              * is updated. In that case, doUnlock will be added to the queue to run after
 681              * this, which will delete the record, as desired. In addition, grant() will
 682              * not do anything, because the lock state will have been set to UNAVAILABLE
 683              * by free().
 684              */
 685
 686             logger.debug("doExtend {}", this);
 687             try (Connection conn = feature.dataSource.getConnection()) {
 688                 /*
 689                  * invoker may have called extend() before free() had a chance to insert
 690                  * the record, thus we have to try to insert, if the update fails
 691                  */
 692                 if (doDbUpdate(conn) || doDbInsert(conn)) {
 693                     grant(true);
 694                     return;
 695                 }
 696             }
 697
 698             removeFromMap();
 699         }
 700
 701         /**
 702          * Inserts the lock into the DB.
 703          *
 704          * @param conn DB connection
 705          * @return {@code true} if a record was successfully inserted, {@code false}
 706          *         otherwise
 707          * @throws SQLException if a DB error occurs
 708          */
 709         protected boolean doDbInsert(Connection conn) throws SQLException {
 710             logger.debug("insert lock record {}", this);
 711             try (PreparedStatement stmt =
 712                             conn.prepareStatement("INSERT INTO pooling.locks (resourceId, host, owner, expirationTime) "
 713                                             + "values (?, ?, ?, timestampadd(second, ?, now()))")) {
 714
 715                 stmt.setString(1, getResourceId());
 716                 stmt.setString(2, feature.hostName);
 717                 stmt.setString(3, feature.uuidString);
 718                 stmt.setInt(4, getHoldSec());
 719
 720                 stmt.executeUpdate();
 721
 722                 this.hostName = feature.hostName;
 723                 this.uuidString = feature.uuidString;
 724
 725                 return true;
 726             }
 727         }
 728
 729         /**
 730          * Updates the lock in the DB.
 731          *
 732          * @param conn DB connection
 733          * @return {@code true} if a record was successfully updated, {@code false}
 734          *         otherwise
 735          * @throws SQLException if a DB error occurs
 736          */
 737         protected boolean doDbUpdate(Connection conn) throws SQLException {
 738             logger.debug("update lock record {}", this);
 739             try (PreparedStatement stmt =
 740                             conn.prepareStatement("UPDATE pooling.locks SET resourceId=?, host=?, owner=?,"
 741                                             + " expirationTime=timestampadd(second, ?, now()) WHERE resourceId=?"
 742                                             + " AND ((host=? AND owner=?) OR expirationTime < now())")) {
 743
 744                 stmt.setString(1, getResourceId());
 745                 stmt.setString(2, feature.hostName);
 746                 stmt.setString(3, feature.uuidString);
 747                 stmt.setInt(4, getHoldSec());
 748
 749                 stmt.setString(5, getResourceId());
 750                 stmt.setString(6, this.hostName);
 751                 stmt.setString(7, this.uuidString);
 752
 753                 if (stmt.executeUpdate() != 1) {
 754                     return false;
 755                 }
 756
 757                 this.hostName = feature.hostName;
 758                 this.uuidString = feature.uuidString;
 759
 760                 return true;
 761             }
 762         }
 763
 764         /**
 765          * Deletes the lock from the DB.
 766          *
 767          * @param conn DB connection
 768          * @throws SQLException if a DB error occurs
 769          */
 770         protected void doDbDelete(Connection conn) throws SQLException {
 771             logger.debug("delete lock record {}", this);
 772             try (PreparedStatement stmt = conn
 773                             .prepareStatement("DELETE FROM pooling.locks WHERE resourceId=? AND host=? AND owner=?")) {
 774
 775                 stmt.setString(1, getResourceId());
 776                 stmt.setString(2, this.hostName);
 777                 stmt.setString(3, this.uuidString);
 778
 779                 stmt.executeUpdate();
 780             }
 781         }
 782
 783         /**
 784          * Removes the lock from the map, and sends a notification using the current
 785          * thread.
 786          */
 787         private void removeFromMap() {
 788             logger.debug("remove lock from map {}", this);
 789             feature.resource2lock.remove(getResourceId(), this);
 790
 791             synchronized (this) {
 792                 if (!isUnavailable()) {
 793                     deny(LOCK_LOST_MSG, true);
 794                 }
 795             }
 796         }
 797
 798         @Override
 799         public String toString() {
 800             return "DistributedLock [state=" + getState() + ", resourceId=" + getResourceId() + ", ownerKey="
 801                             + getOwnerKey() + ", holdSec=" + getHoldSec() + ", hostName=" + hostName + ", uuidString="
 802                             + uuidString + "]";
 803         }
 804     }
 805
 806     @FunctionalInterface
 807     private static interface RunnableWithEx {
 808         void run() throws SQLException;
 809     }
 810
 811     // these may be overridden by junit tests
 812
 813     protected Properties getProperties(String fileName) {
 814         return SystemPersistenceConstants.getManager().getProperties(fileName);
 815     }
 816
 817     protected DistributedLock makeLock(LockState state, String resourceId, String ownerKey, int holdSec,
 818                     LockCallback callback) {
 819         return new DistributedLock(state, resourceId, ownerKey, holdSec, callback, this);
 820     }
 821 }