Coverage Report

Coverage Report - liquibase.lockservice.LockService

Classes in this File

Line Coverage

Branch Coverage

Complexity

LockService

80%

82/102

73%

28/38

 package liquibase.lockservice;
 
 import liquibase.database.Database;
 import liquibase.exception.DatabaseException;
 import liquibase.exception.LockException;
 import liquibase.executor.Executor;
 import liquibase.executor.ExecutorService;
 import liquibase.logging.LogFactory;
 import liquibase.statement.SqlStatement;
 import liquibase.statement.core.LockDatabaseChangeLogStatement;
 import liquibase.statement.core.SelectFromDatabaseChangeLogLockStatement;
 import liquibase.statement.core.UnlockDatabaseChangeLogStatement;
 import liquibase.statement.core.RawSqlStatement;
 
 import java.text.DateFormat;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class LockService {
 
     private Database database;
 
     private boolean hasChangeLogLock = false;
 
     private long changeLogLockWaitTime = 1000 * 60 * 5; // default to 5 mins
     private long changeLogLocRecheckTime = 1000 * 10; // default to every 10 seconds
 
     private static Map<Database, LockService> instances = new ConcurrentHashMap<Database, LockService>();
 
     private LockService(Database database) {
         this.database = database;
     }
 
     public static LockService getInstance(Database database) {
         if (!instances.containsKey(database)) {
             instances.put(database, new LockService(database));
         }
         return instances.get(database);
     }
 
     public void setChangeLogLockWaitTime(long changeLogLockWaitTime) {
         this.changeLogLockWaitTime = changeLogLockWaitTime;
     }
 
     public void setChangeLogLockRecheckTime(long changeLogLocRecheckTime) {
         this.changeLogLocRecheckTime = changeLogLocRecheckTime;
     }
 
     public boolean hasChangeLogLock() {
         return hasChangeLogLock;
     }
 
     public void waitForLock() throws LockException {
 
         boolean locked = false;
         long timeToGiveUp = new Date().getTime() + changeLogLockWaitTime;
         while (!locked && new Date().getTime() < timeToGiveUp) {
             locked = acquireLock();
             if (!locked) {
                 System.out.println("Waiting for changelog lock....");
                 try {
                     Thread.sleep(changeLogLocRecheckTime);
                 } catch (InterruptedException e) {
                     ;
                 }
             }
         }
 
         if (!locked) {
             DatabaseChangeLogLock[] locks = listLocks();
             String lockedBy;
             if (locks.length > 0) {
                 DatabaseChangeLogLock lock = locks[0];
                 lockedBy = lock.getLockedBy()
                         + " since "
                         + DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.SHORT).format(
                                 lock.getLockGranted());
             } else {
                 lockedBy = "UNKNOWN";
             }
             throw new LockException("Could not acquire change log lock.  Currently locked by " + lockedBy);
         }
     }
 
     public boolean acquireLock() throws LockException {
         if (hasChangeLogLock) {
             return true;
         }
 
         Executor executor = ExecutorService.getInstance().getExecutor(database);
 
         try {
             database.rollback();
             database.checkDatabaseChangeLogLockTable();
 
             Boolean locked = (Boolean) ExecutorService.getInstance().getExecutor(database)
                     .queryForObject(new SelectFromDatabaseChangeLogLockStatement("LOCKED"), Boolean.class);
 
             if (locked) {
                 return false;
             } else {
 
                 executor.comment("Lock Database");
                 int rowsUpdated = executor.update(new LockDatabaseChangeLogStatement());
                 if (rowsUpdated > 1) {
                     throw new LockException("Did not update change log lock correctly");
                 }
                 if (rowsUpdated == 0) {
                     // another node was faster
                     return false;
                 }
                 database.commit();
                 LogFactory.getLogger().info("Successfully acquired change log lock");
 
                 hasChangeLogLock = true;
 
                 database.setCanCacheLiquibaseTableInfo(true);
                 return true;
             }
         } catch (Exception e) {
             throw new LockException(e);
         } finally {
             try {
                 database.rollback();
             } catch (DatabaseException e) {
                 ;
             }
         }
 
     }
 
     public void releaseLock() throws LockException {
         Executor executor = ExecutorService.getInstance().getExecutor(database);
         try {
             if (database.hasDatabaseChangeLogLockTable()) {
                 executor.comment("Release Database Lock");
                 database.rollback();
                 int updatedRows = executor.update(new UnlockDatabaseChangeLogStatement());
                 if (updatedRows != 1) {
                     throw new LockException("Did not update change log lock correctly.\n\n"
                             + updatedRows
                             + " rows were updated instead of the expected 1 row using executor "
                             + executor.getClass().getName()
                             + " there are "
                             + executor.queryForInt(new RawSqlStatement("select count(*) from "
                                     + database.getDatabaseChangeLogLockTableName())) + " rows in the table");
                 }
                 database.commit();
                 hasChangeLogLock = false;
 
                 instances.remove(this.database);
 
                 database.setCanCacheLiquibaseTableInfo(false);
 
                 LogFactory.getLogger().info("Successfully released change log lock");
             }
         } catch (Exception e) {
             throw new LockException(e);
         } finally {
             try {
                 database.rollback();
             } catch (DatabaseException e) {
                 ;
             }
         }
     }
 
     public DatabaseChangeLogLock[] listLocks() throws LockException {
         try {
             if (!database.hasDatabaseChangeLogLockTable()) {
                 return new DatabaseChangeLogLock[0];
             }
 
             List<DatabaseChangeLogLock> allLocks = new ArrayList<DatabaseChangeLogLock>();
             SqlStatement sqlStatement = new SelectFromDatabaseChangeLogLockStatement("ID", "LOCKED", "LOCKGRANTED",
                     "LOCKEDBY");
             List<Map> rows = ExecutorService.getInstance().getExecutor(database).queryForList(sqlStatement);
             for (Map columnMap : rows) {
                 Object lockedValue = columnMap.get("LOCKED");
                 Boolean locked;
                 if (lockedValue instanceof Number) {
                     locked = ((Number) lockedValue).intValue() == 1;
                 } else {
                     locked = (Boolean) lockedValue;
                 }
                 if (locked != null && locked) {
                     allLocks.add(new DatabaseChangeLogLock(((Number) columnMap.get("ID")).intValue(), (Date) columnMap
                             .get("LOCKGRANTED"), (String) columnMap.get("LOCKEDBY")));
                 }
             }
             return allLocks.toArray(new DatabaseChangeLogLock[allLocks.size()]);
         } catch (Exception e) {
             throw new LockException(e);
         }
     }
 
     /**
      * Releases whatever locks are on the database change log table
      */
     public void forceReleaseLock() throws LockException, DatabaseException {
         database.checkDatabaseChangeLogLockTable();
         releaseLock();
         /*
          * try { releaseLock(); } catch (LockException e) { // ignore ?
          * LogFactory.getLogger().info("Ignored exception in forceReleaseLock: " + e.getMessage()); }
          */
     }
 
     /**
      * Clears information the lock handler knows about the tables. Should only be called by Liquibase internal calls
      */
     public void reset() {
         hasChangeLogLock = false;
     }
 
     public static void resetAll() {
         for (Map.Entry<Database, LockService> entity : instances.entrySet()) {
             entity.getValue().reset();
         }
     }
 }

1		package liquibase.lockservice;
2
3		import liquibase.database.Database;
4		import liquibase.exception.DatabaseException;
5		import liquibase.exception.LockException;
6		import liquibase.executor.Executor;
7		import liquibase.executor.ExecutorService;
8		import liquibase.logging.LogFactory;
9		import liquibase.statement.SqlStatement;
10		import liquibase.statement.core.LockDatabaseChangeLogStatement;
11		import liquibase.statement.core.SelectFromDatabaseChangeLogLockStatement;
12		import liquibase.statement.core.UnlockDatabaseChangeLogStatement;
13		import liquibase.statement.core.RawSqlStatement;
14
15		import java.text.DateFormat;
16		import java.util.ArrayList;
17		import java.util.Date;
18		import java.util.List;
19		import java.util.Map;
20		import java.util.concurrent.ConcurrentHashMap;
21
22		public class LockService {
23
24		private Database database;
25
26	13	private boolean hasChangeLogLock = false;
27
28	13	private long changeLogLockWaitTime = 1000 * 60 * 5; // default to 5 mins
29	13	private long changeLogLocRecheckTime = 1000 * 10; // default to every 10 seconds
30
31	1	private static Map<Database, LockService> instances = new ConcurrentHashMap<Database, LockService>();
32
33	13	private LockService(Database database) {
34	13	this.database = database;
35	13	}
36
37		public static LockService getInstance(Database database) {
38	23	if (!instances.containsKey(database)) {
39	13	instances.put(database, new LockService(database));
40		}
41	23	return instances.get(database);
42		}
43
44		public void setChangeLogLockWaitTime(long changeLogLockWaitTime) {
45	1	this.changeLogLockWaitTime = changeLogLockWaitTime;
46	1	}
47
48		public void setChangeLogLockRecheckTime(long changeLogLocRecheckTime) {
49	2	this.changeLogLocRecheckTime = changeLogLocRecheckTime;
50	2	}
51
52		public boolean hasChangeLogLock() {
53	2	return hasChangeLogLock;
54		}
55
56		public void waitForLock() throws LockException {
57
58	3	boolean locked = false;
59	3	long timeToGiveUp = new Date().getTime() + changeLogLockWaitTime;
60	10	while (!locked && new Date().getTime() < timeToGiveUp) {
61	7	locked = acquireLock();
62	7	if (!locked) {
63	5	System.out.println("Waiting for changelog lock....");
64		try {
65	5	Thread.sleep(changeLogLocRecheckTime);
66	0	} catch (InterruptedException e) {
67		;
68	5	}
69		}
70		}
71
72	3	if (!locked) {
73	1	DatabaseChangeLogLock[] locks = listLocks();
74		String lockedBy;
75	1	if (locks.length > 0) {
76	1	DatabaseChangeLogLock lock = locks[0];
77	1	lockedBy = lock.getLockedBy()
78		+ " since "
79		+ DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.SHORT).format(
80		lock.getLockGranted());
81	1	} else {
82	0	lockedBy = "UNKNOWN";
83		}
84	1	throw new LockException("Could not acquire change log lock. Currently locked by " + lockedBy);
85		}
86	2	}
87
88		public boolean acquireLock() throws LockException {
89	10	if (hasChangeLogLock) {
90	1	return true;
91		}
92
93	9	Executor executor = ExecutorService.getInstance().getExecutor(database);
94
95		try {
96	9	database.rollback();
97	9	database.checkDatabaseChangeLogLockTable();
98
99	9	Boolean locked = (Boolean) ExecutorService.getInstance().getExecutor(database)
100		.queryForObject(new SelectFromDatabaseChangeLogLockStatement("LOCKED"), Boolean.class);
101
102	9	if (locked) {
103	6	return false;
104		} else {
105
106	3	executor.comment("Lock Database");
107	3	int rowsUpdated = executor.update(new LockDatabaseChangeLogStatement());
108	3	if (rowsUpdated > 1) {
109	0	throw new LockException("Did not update change log lock correctly");
110		}
111	3	if (rowsUpdated == 0) {
112		// another node was faster
113	0	return false;
114		}
115	3	database.commit();
116	3	LogFactory.getLogger().info("Successfully acquired change log lock");
117
118	3	hasChangeLogLock = true;
119
120	3	database.setCanCacheLiquibaseTableInfo(true);
121	3	return true;
122		}
123	0	} catch (Exception e) {
124	0	throw new LockException(e);
125		} finally {
126	0	try {
127	9	database.rollback();
128	0	} catch (DatabaseException e) {
129		;
130	18	}
131		}
132
133		}
134
135		public void releaseLock() throws LockException {
136	1	Executor executor = ExecutorService.getInstance().getExecutor(database);
137		try {
138	1	if (database.hasDatabaseChangeLogLockTable()) {
139	1	executor.comment("Release Database Lock");
140	1	database.rollback();
141	1	int updatedRows = executor.update(new UnlockDatabaseChangeLogStatement());
142	1	if (updatedRows != 1) {
143	0	throw new LockException("Did not update change log lock correctly.\n\n"
144		+ updatedRows
145		+ " rows were updated instead of the expected 1 row using executor "
146		+ executor.getClass().getName()
147		+ " there are "
148		+ executor.queryForInt(new RawSqlStatement("select count(*) from "
149		+ database.getDatabaseChangeLogLockTableName())) + " rows in the table");
150		}
151	1	database.commit();
152	1	hasChangeLogLock = false;
153
154	1	instances.remove(this.database);
155
156	1	database.setCanCacheLiquibaseTableInfo(false);
157
158	1	LogFactory.getLogger().info("Successfully released change log lock");
159		}
160	0	} catch (Exception e) {
161	0	throw new LockException(e);
162		} finally {
163	0	try {
164	1	database.rollback();
165	0	} catch (DatabaseException e) {
166		;
167	1	}
168	0	}
169	1	}
170
171		public DatabaseChangeLogLock[] listLocks() throws LockException {
172		try {
173	4	if (!database.hasDatabaseChangeLogLockTable()) {
174	1	return new DatabaseChangeLogLock[0];
175		}
176
177	3	List<DatabaseChangeLogLock> allLocks = new ArrayList<DatabaseChangeLogLock>();
178	3	SqlStatement sqlStatement = new SelectFromDatabaseChangeLogLockStatement("ID", "LOCKED", "LOCKGRANTED",
179		"LOCKEDBY");
180	3	List<Map> rows = ExecutorService.getInstance().getExecutor(database).queryForList(sqlStatement);
181	3	for (Map columnMap : rows) {
182	2	Object lockedValue = columnMap.get("LOCKED");
183		Boolean locked;
184	2	if (lockedValue instanceof Number) {
185	0	locked = ((Number) lockedValue).intValue() == 1;
186		} else {
187	2	locked = (Boolean) lockedValue;
188		}
189	2	if (locked != null && locked) {
190	2	allLocks.add(new DatabaseChangeLogLock(((Number) columnMap.get("ID")).intValue(), (Date) columnMap
191		.get("LOCKGRANTED"), (String) columnMap.get("LOCKEDBY")));
192		}
193	2	}
194	3	return allLocks.toArray(new DatabaseChangeLogLock[allLocks.size()]);
195	0	} catch (Exception e) {
196	0	throw new LockException(e);
197		}
198		}
199
200		/**
201		* Releases whatever locks are on the database change log table
202		*/
203		public void forceReleaseLock() throws LockException, DatabaseException {
204	0	database.checkDatabaseChangeLogLockTable();
205	0	releaseLock();
206		/*
207		* try { releaseLock(); } catch (LockException e) { // ignore ?
208		* LogFactory.getLogger().info("Ignored exception in forceReleaseLock: " + e.getMessage()); }
209		*/
210	0	}
211
212		/**
213		* Clears information the lock handler knows about the tables. Should only be called by Liquibase internal calls
214		*/
215		public void reset() {
216	156	hasChangeLogLock = false;
217	156	}
218
219		public static void resetAll() {
220	22	for (Map.Entry<Database, LockService> entity : instances.entrySet()) {
221	156	entity.getValue().reset();
222		}
223	22	}
224		}