GC: JDBCPersistenceManagerImpl - com.ibm.jbatch.container.services.impl.JDBCPersistenceManagerImpl (.java)

Maven-Central
/
com.ibm.jbatch / c.....batch-runtime-all
/
1.0-b27

1.0
1.0-b29
1.0-b28
1.0-b27
1.0-b26
1.0-b25
1.0-b24
1.0-b23
1.0-b22
1.0-b21
1.0-b20
1.0-b19
1.0-b18
1.0-b17
1.0-b16
1.0-b15
1.0-b14
1.0-b13
/
com.ibm.jbatch.container.services.impl.JDBCPersistenceManagerImpl
Find Usages
Diff
Raw
Download
HTML Widget
Stack Overflow Questions
1
   /*
2
    * Copyright 2012 International Business Machines Corp.
3
    * 
4
    * See the NOTICE file distributed with this work for additional information
5
    * regarding copyright ownership. Licensed under the Apache License, 
6
    * Version 2.0 (the "License"); you may not use this file except in compliance
7
    * with the License. You may obtain a copy of the License at
8
    * 
9
    *   http://www.apache.org/licenses/LICENSE-2.0
10
   * 
11
   * Unless required by applicable law or agreed to in writing, software
12
   * distributed under the License is distributed on an "AS IS" BASIS,
13
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14
   * See the License for the specific language governing permissions and
15
   * limitations under the License.
16
   */
17
  package com.ibm.jbatch.container.services.impl;
18
  
19
  import java.io.ByteArrayInputStream;
20
  import java.io.ByteArrayOutputStream;
21
  import java.io.IOException;
22
  import java.io.ObjectInputStream;
23
  import java.io.ObjectOutputStream;
24
  import java.io.PrintWriter;
25
  import java.io.Serializable;
26
  import java.io.StringWriter;
27
  import java.sql.Connection;
28
  import java.sql.DatabaseMetaData;
29
  import java.sql.DriverManager;
30
  import java.sql.PreparedStatement;
31
  import java.sql.ResultSet;
32
  import java.sql.SQLException;
33
  import java.sql.Statement;
34
  import java.sql.Timestamp;
35
  import java.util.ArrayList;
36
  import java.util.HashMap;
37
  import java.util.HashSet;
38
  import java.util.List;
39
  import java.util.Map;
40
  import java.util.Properties;
41
  import java.util.Set;
42
  import java.util.logging.Level;
43
  import java.util.logging.Logger;
44
  
45
  import javax.batch.operations.NoSuchJobExecutionException;
46
  import javax.batch.runtime.BatchStatus;
47
  import javax.batch.runtime.JobInstance;
48
  import javax.batch.runtime.Metric;
49
  import javax.batch.runtime.StepExecution;
50
  import javax.naming.Context;
51
  import javax.naming.InitialContext;
52
  import javax.naming.NamingException;
53
  import javax.sql.DataSource;
54
  
55
  import com.ibm.jbatch.container.context.impl.MetricImpl;
56
  import com.ibm.jbatch.container.context.impl.StepContextImpl;
57
  import com.ibm.jbatch.container.exception.BatchContainerServiceException;
58
  import com.ibm.jbatch.container.exception.PersistenceException;
59
  import com.ibm.jbatch.container.impl.PartitionedStepBuilder;
60
  import com.ibm.jbatch.container.jobinstance.JobInstanceImpl;
61
  import com.ibm.jbatch.container.jobinstance.JobOperatorJobExecution;
62
  import com.ibm.jbatch.container.jobinstance.RuntimeFlowInSplitExecution;
63
  import com.ibm.jbatch.container.jobinstance.RuntimeJobExecution;
64
  import com.ibm.jbatch.container.jobinstance.StepExecutionImpl;
65
  import com.ibm.jbatch.container.persistence.CheckpointData;
66
  import com.ibm.jbatch.container.persistence.CheckpointDataKey;
67
  import com.ibm.jbatch.container.services.IJobExecution;
68
  import com.ibm.jbatch.container.services.IPersistenceManagerService;
69
  import com.ibm.jbatch.container.status.JobStatus;
70
  import com.ibm.jbatch.container.status.StepStatus;
71
  import com.ibm.jbatch.container.util.TCCLObjectInputStream;
72
  import com.ibm.jbatch.spi.services.IBatchConfig;
73
  
74
  public class JDBCPersistenceManagerImpl implements IPersistenceManagerService, JDBCPersistenceManagerSQLConstants {
75
  
76
  	private static final String CLASSNAME = JDBCPersistenceManagerImpl.class.getName();
77
  
78
  	private final static Logger logger = Logger.getLogger(CLASSNAME);
79
  
80
  	private IBatchConfig batchConfig = null;
81
  
82
  	protected DataSource dataSource = null;
83
  	protected String jndiName = null;
84
  
85
  	protected String driver = ""; 
86
  	protected String schema = "";
87
  	protected String url = ""; 
88
  	protected String userId = "";
89
  	protected String pwd = "";
90
  
91
  	/* (non-Javadoc)
92
  	 * @see com.ibm.jbatch.container.services.impl.AbstractPersistenceManagerImpl#init(com.ibm.jbatch.container.IBatchConfig)
93
  	 */
94
  	@Override
95
  	public void init(IBatchConfig batchConfig) throws BatchContainerServiceException {
96
  		logger.config("Entering CLASSNAME.init(), batchConfig =" + batchConfig);
97
  
98
  		this.batchConfig = batchConfig;
99
  
100
 		schema = batchConfig.getDatabaseConfigurationBean().getSchema();
101
 
102
 		if (!batchConfig.isJ2seMode()) {
103
 			jndiName = batchConfig.getDatabaseConfigurationBean().getJndiName();
104
 
105
 			logger.config("JNDI name = " + jndiName);
106
 
107
 			if (jndiName == null || jndiName.equals("")) {
108
 				throw new BatchContainerServiceException("JNDI name is not defined.");
109
 			}
110
 
111
 			try {
112
 				Context ctx = new InitialContext();
113
 				dataSource = (DataSource) ctx.lookup(jndiName);
114
 
115
 			} catch (NamingException e) {
116
 				logger.severe("Lookup failed for JNDI name: " + jndiName + 
117
 						".  One cause of this could be that the batch runtime is incorrectly configured to EE mode when it should be in SE mode.");
118
 				throw new BatchContainerServiceException(e);
119
 			}
120
 
121
 		} else {
122
 			driver = batchConfig.getDatabaseConfigurationBean().getJdbcDriver();
123
 			url = batchConfig.getDatabaseConfigurationBean().getJdbcUrl();
124
 			userId = batchConfig.getDatabaseConfigurationBean().getDbUser();
125
 			pwd = batchConfig.getDatabaseConfigurationBean().getDbPassword();
126
 
127
 			logger.config("driver: " + driver + ", url: " + url);
128
 		}
129
 
130
 		try {
131
 			// only auto-create on Derby
132
 			if(isDerby()) {	
133
 				if(!isSchemaValid()) {
134
 					createSchema();
135
 				}
136
 				checkAllTables();
137
 			}
138
 		} catch (SQLException e) {
139
 			logger.severe(e.getLocalizedMessage());
140
 			throw new BatchContainerServiceException(e);
141
 		}
142
 
143
 		logger.config("Exiting CLASSNAME.init()");
144
 	}

	
 Checks if the default schema JBATCH or the schema defined in batch-config exists.
 
 
 Returns:true if the schema exists, false otherwise. Throws:java.sql.SQLException
151
 
152
 	private boolean isSchemaValid() throws SQLException {
153
 		logger.entering(CLASSNAME, "isSchemaValid");
154
 		Connection conn = getConnectionToDefaultSchema();
155
 		DatabaseMetaData dbmd = conn.getMetaData();
156
 		ResultSet rs = dbmd.getSchemas();
157
 		while(rs.next()) {
158
 			if (schema.equalsIgnoreCase(rs.getString("TABLE_SCHEM")) ) {
159
 				cleanupConnection(conn, rs, null);
160
 				logger.exiting(CLASSNAME, "isSchemaValid", true);
161
 				return true;
162
 			}
163
 		}
164
 		cleanupConnection(conn, rs, null);
165
 		logger.exiting(CLASSNAME, "isSchemaValid", false);
166
 		return false;
167
 	}
168
 
169
 	private boolean isDerby() throws SQLException {
170
 		logger.entering(CLASSNAME, "isDerby");
171
 		Connection conn = getConnectionToDefaultSchema();
172
 		DatabaseMetaData dbmd = conn.getMetaData();
173
 		boolean derby = dbmd.getDatabaseProductName().toLowerCase().indexOf("derby") > 0;
174
 		logger.exiting(CLASSNAME, "isDerby", derby);
175
 		return derby;
176
 	}

	
 Creates the default schema JBATCH or the schema defined in batch-config.
 
 
 Throws:java.sql.SQLException
182
 
183
 	private void createSchema() throws SQLException {
184
 		logger.entering(CLASSNAME, "createSchema");
185
 		Connection conn = getConnectionToDefaultSchema();
186
 
187
 		logger.log(Level.INFO, schema + " schema does not exists. Trying to create it.");
188
 		PreparedStatement ps = null;
189
 		ps = conn.prepareStatement("CREATE SCHEMA " + schema);
190
 		ps.execute();
191
 
192
 		cleanupConnection(conn, null, ps);
193
 		logger.exiting(CLASSNAME, "createSchema");
194
 	}

	
 Checks if all the runtime batch table exists. If not, it creates them.
  
 
 Throws:java.sql.SQLException
200
 
201
 	private void checkAllTables() throws SQLException {
202
 		logger.entering(CLASSNAME, "checkAllTables");
203
 
204
 		createIfNotExists(CHECKPOINTDATA_TABLE, CREATE_TAB_CHECKPOINTDATA);
205
 		executeStatement(CREATE_CHECKPOINTDATA_INDEX);
206
 		createIfNotExists(JOBINSTANCEDATA_TABLE, CREATE_TAB_JOBINSTANCEDATA);
207
 
208
 		createIfNotExists(EXECUTIONINSTANCEDATA_TABLE,
209
 				CREATE_TAB_EXECUTIONINSTANCEDATA);
210
 		createIfNotExists(STEPEXECUTIONINSTANCEDATA_TABLE,
211
 				CREATE_TAB_STEPEXECUTIONINSTANCEDATA);
212
 
213
 		createIfNotExists(JOBSTATUS_TABLE, CREATE_TAB_JOBSTATUS);
214
 		createIfNotExists(STEPSTATUS_TABLE, CREATE_TAB_STEPSTATUS);	
215
 
216
 		logger.exiting(CLASSNAME, "checkAllTables");
217
 	}

	
 Creates tableName using the createTableStatement DDL.
 
 
 Parameters:tableNamecreateTableStatement Throws:java.sql.SQLException
225
 
226
 	private void createIfNotExists(String tableName, String createTableStatement) throws SQLException {
227
 		logger.entering(CLASSNAME, "createIfNotExists", new Object[] {tableName, createTableStatement});
228
 
229
 		Connection conn = getConnection();
230
 		DatabaseMetaData dbmd = conn.getMetaData();
231
 		ResultSet rs = dbmd.getTables(null, schema, tableName, null);
232
 		PreparedStatement ps = null;
233
 		if(!rs.next()) {
234
 			logger.log(Level.INFO, tableName + " table does not exists. Trying to create it.");
235
 			ps = conn.prepareStatement(createTableStatement);
236
 			ps.executeUpdate();
237
 		}
238
 
239
 		cleanupConnection(conn, rs, ps);
240
 		logger.exiting(CLASSNAME, "createIfNotExists");
241
 	}

	
 Executes the provided SQL statement
 
 
 Parameters:statement Throws:java.sql.SQLException
248
 
249
 	private void executeStatement(String statement) throws SQLException {
250
 		logger.entering(CLASSNAME, "executeStatement", statement);
251
 
252
 		Connection conn = getConnection();
253
 		PreparedStatement ps = null;
254
 
255
 		ps = conn.prepareStatement(statement);
256
 		ps.executeUpdate();
257
 
258
 		cleanupConnection(conn, ps);
259
 		logger.exiting(CLASSNAME, "executeStatement");
260
 	}
261
 
262
 
263
 	/* (non-Javadoc)
264
 	 * @see com.ibm.jbatch.container.services.impl.AbstractPersistenceManagerImpl#createCheckpointData(com.ibm.ws.batch.container.checkpoint.CheckpointDataKey, com.ibm.ws.batch.container.checkpoint.CheckpointData)
265
 	 */
266
 	@Override
267
 	public void createCheckpointData(CheckpointDataKey key, CheckpointData value) {
268
 		logger.entering(CLASSNAME, "createCheckpointData", new Object[] {key, value});
269
 		insertCheckpointData(key.getCommaSeparatedKey(), value);
270
 		logger.exiting(CLASSNAME, "createCheckpointData");
271
 	}
272
 
273
 	/* (non-Javadoc)
274
 	 * @see com.ibm.jbatch.container.services.impl.AbstractPersistenceManagerImpl#getCheckpointData(com.ibm.ws.batch.container.checkpoint.CheckpointDataKey)
275
 	 */
276
 	@Override
277
 	public CheckpointData getCheckpointData(CheckpointDataKey key) {
278
 		logger.entering(CLASSNAME, "getCheckpointData", key==null ? "<null>" : key);
279
 		CheckpointData checkpointData = queryCheckpointData(key.getCommaSeparatedKey());
280
 		logger.exiting(CLASSNAME, "getCheckpointData", checkpointData==null ? "<null>" : checkpointData);
281
 		return checkpointData;
282
 	}
283
 
284
 	/* (non-Javadoc)
285
 	 * @see com.ibm.jbatch.container.services.impl.AbstractPersistenceManagerImpl#updateCheckpointData(com.ibm.ws.batch.container.checkpoint.CheckpointDataKey, com.ibm.ws.batch.container.checkpoint.CheckpointData)
286
 	 */
287
 	@Override
288
 	public void updateCheckpointData(CheckpointDataKey key, CheckpointData value) {
289
 		logger.entering(CLASSNAME, "updateCheckpointData", new Object[] {key, value});
290
 		CheckpointData data = queryCheckpointData(key.getCommaSeparatedKey());
291
 		if(data != null) {
292
 			updateCheckpointData(key.getCommaSeparatedKey(), value);
293
 		} else {
294
 			createCheckpointData(key, value);
295
 		}
296
 		logger.exiting(CLASSNAME, "updateCheckpointData");
297
 	}


	
 Returns:the database connection and sets it to the default schema JBATCH or the schema defined in batch-config. Throws:java.sql.SQLException
304
 
305
 	protected Connection getConnection() throws SQLException {
306
 		logger.finest("Entering: " + CLASSNAME + ".getConnection");
307
 		Connection connection = null;
308
 
309
 		if(!batchConfig.isJ2seMode()) {
310
 			logger.finest("J2EE mode, getting connection from data source");
311
 			connection = dataSource.getConnection();
312
 			logger.finest("autocommit="+connection.getAutoCommit());
313
 		} else {
314
 			try {
315
 				Class.forName(driver);
316
 			} catch (ClassNotFoundException e) {
317
 				throw new PersistenceException(e);
318
 			}
319
 			logger.finest("JSE mode, getting connection from " + url);
320
 			connection = DriverManager.getConnection(url, userId, pwd);
321
 			logger.finest("autocommit="+connection.getAutoCommit());
322
 		}
323
 		setSchemaOnConnection(connection);
324
 
325
 		logger.finest("Exiting: " + CLASSNAME + ".getConnection() with conn =" + connection);
326
 		return connection;
327
 	}

	
 Returns:the database connection. The schema is set to whatever default its used by the underlying database. Throws:java.sql.SQLException
332
 
333
 	protected Connection getConnectionToDefaultSchema() throws SQLException {
334
 		logger.finest("Entering getConnectionToDefaultSchema");
335
 		Connection connection = null;
336
 
337
 		if(!batchConfig.isJ2seMode()) {
338
 			logger.finest("J2EE mode, getting connection from data source");
339
 			try {
340
 				connection = dataSource.getConnection();
341
 			} catch(SQLException e) {
342
 				logException("FAILED GETTING DATABASE CONNECTION", e);
343
 				throw new PersistenceException(e);
344
 			}
345
 			logger.finest("autocommit="+connection.getAutoCommit());
346
 		} else {
347
 			try {
348
 				Class.forName(driver);
349
 			} catch (ClassNotFoundException e) {
350
 				logException("ClassNotFoundException: Cannot load driver class: " + driver, e);
351
 				throw new PersistenceException(e);
352
 			}
353
 			logger.finest("JSE mode, getting connection from " + url);
354
 			try {
355
 				connection = DriverManager.getConnection(url, userId, pwd);
356
 			} catch (SQLException e) {
357
 				logException("FAILED GETTING DATABASE CONNECTION.  FOR EMBEDDED DERBY CHECK FOR OTHER USER LOCKING THE CURRENT DATABASE (Try using a different database instance).", e); 
358
 				throw new PersistenceException(e);
359
 			}
360
 			logger.finest("autocommit="+connection.getAutoCommit());
361
 		}
362
 		logger.finest("Exiting from getConnectionToDefaultSchema, conn= " +connection);
363
 		return connection;
364
 	}
365
 
366
 	private void logException(String msg, Exception e) {
367
 		StringWriter sw = new StringWriter();
368
 		PrintWriter pw = new PrintWriter(sw);
369
 		e.printStackTrace(pw);
370
 
371
 		logger.log(Level.SEVERE, msg +  "; Exception stack trace: " + sw);
372
 	}

	
 Set the default schema JBATCH or the schema defined in batch-config on the connection object.
 
 
 Parameters:connection Throws:java.sql.SQLException
379
 
380
 	private void setSchemaOnConnection(Connection connection) throws SQLException {
381
 		logger.finest("Entering " + CLASSNAME +".setSchemaOnConnection()");
382
 
383
 		PreparedStatement ps = null;
384
 		ps = connection.prepareStatement("SET SCHEMA ?");
385
 		ps.setString(1, schema);
386
 		ps.executeUpdate(); 
387
 
388
 		ps.close();
389
 
390
 		logger.finest("Exiting " + CLASSNAME +".setSchemaOnConnection()");
391
 	}

	
 select data from DB table
 
 
 Parameters:key - the IPersistenceDataKey object Returns:List of serializable objects store in the DB table
 
 Ex. select id, obj from tablename where id = ?
400
 
401
 	private CheckpointData queryCheckpointData(Object key) {
402
 		logger.entering(CLASSNAME, "queryCheckpointData", new Object[] {key, SELECT_CHECKPOINTDATA});
403
 		Connection conn = null;
404
 		PreparedStatement statement = null;
405
 		ResultSet rs = null;
406
 		ObjectInputStream objectIn = null;
407
 		CheckpointData data = null;
408
 		try {
409
 			conn = getConnection();
410
 			statement = conn.prepareStatement(SELECT_CHECKPOINTDATA);
411
 			statement.setObject(1, key);
412
 			rs = statement.executeQuery();
413
 			if (rs.next()) {
414
 				byte[] buf = rs.getBytes("obj");
415
 				data = (CheckpointData)deserializeObject(buf);
416
 			}
417
 		} catch (SQLException e) {
418
 			throw new PersistenceException(e);
419
 		} catch (IOException e) {
420
 			throw new PersistenceException(e);
421
 		} catch (ClassNotFoundException e) {
422
 			throw new PersistenceException(e);
423
 		} finally {
424
 			if (objectIn != null) {
425
 				try {
426
 					objectIn.close();
427
 				} catch (IOException e) {
428
 					throw new PersistenceException(e);
429
 				}
430
 			}
431
 			cleanupConnection(conn, rs, statement);
432
 		}
433
 		logger.exiting(CLASSNAME, "queryCheckpointData");
434
 		return data;	
435
 	}


	
 insert data to DB table
 
 
 Parameters:key - the IPersistenceDataKey objectvalue - serializable object to store  
 
 Ex. insert into tablename values(?, ?)
445
 
446
 	private <T> void insertCheckpointData(Object key, T value) {
447
 		logger.entering(CLASSNAME, "insertCheckpointData", new Object[] {key, value});
448
 		Connection conn = null;
449
 		PreparedStatement statement = null;
450
 		ByteArrayOutputStream baos = null;
451
 		ObjectOutputStream oout = null;
452
 		byte[] b;
453
 		try {
454
 			conn = getConnection();
455
 			statement = conn.prepareStatement(INSERT_CHECKPOINTDATA);
456
 			baos = new ByteArrayOutputStream();
457
 			oout = new ObjectOutputStream(baos);
458
 			oout.writeObject(value);
459
 
460
 			b = baos.toByteArray();
461
 
462
 			statement.setObject(1, key);
463
 			statement.setBytes(2, b);
464
 			statement.executeUpdate();
465
 		} catch (SQLException e) {
466
 			throw new PersistenceException(e);
467
 		} catch (IOException e) {
468
 			throw new PersistenceException(e);
469
 		} finally {
470
 			if (baos != null) {
471
 				try {
472
 					baos.close();
473
 				} catch (IOException e) {
474
 					throw new PersistenceException(e);
475
 				}
476
 			}
477
 			if (oout != null) {
478
 				try {
479
 					oout.close();
480
 				} catch (IOException e) {
481
 					throw new PersistenceException(e);
482
 				}
483
 			}
484
 			cleanupConnection(conn, null, statement);
485
 		}
486
 		logger.exiting(CLASSNAME, "insertCheckpointData");
487
 	}

	
 update data in DB table
 
 
 Parameters:value - serializable object to storekey - the IPersistenceDataKey objectquery - SQL statement to execute. 
 
 Ex. update tablename set obj = ? where id = ?
497
 
498
 	private void updateCheckpointData(Object key, CheckpointData value) {
499
 		logger.entering(CLASSNAME, "updateCheckpointData", new Object[] {key, value});
500
 		Connection conn = null;
501
 		PreparedStatement statement = null;
502
 		ByteArrayOutputStream baos = null;
503
 		ObjectOutputStream oout = null;
504
 		byte[] b;
505
 		try {
506
 			conn = getConnection();
507
 			statement = conn.prepareStatement(UPDATE_CHECKPOINTDATA);
508
 			baos = new ByteArrayOutputStream();
509
 			oout = new ObjectOutputStream(baos);
510
 			oout.writeObject(value);
511
 
512
 			b = baos.toByteArray();
513
 
514
 			statement.setBytes(1, b);
515
 			statement.setObject(2, key);
516
 			statement.executeUpdate();
517
 		} catch (SQLException e) {
518
 			throw new PersistenceException(e);
519
 		} catch (IOException e) {
520
 			throw new PersistenceException(e);
521
 		} finally {
522
 			if (baos != null) {
523
 				try {
524
 					baos.close();
525
 				} catch (IOException e) {
526
 					throw new PersistenceException(e);
527
 				}
528
 			}
529
 			if (oout != null) {
530
 				try {
531
 					oout.close();
532
 				} catch (IOException e) {
533
 					throw new PersistenceException(e);
534
 				}
535
 			}
536
 			cleanupConnection(conn, null, statement);
537
 		}
538
 		logger.exiting(CLASSNAME, "updateCheckpointData");
539
 	}



	
 closes connection, result set and statement
 
 
 Parameters:conn - connection object to closers - result set object to closestatement - statement object to close
549
 
550
 	private void cleanupConnection(Connection conn, ResultSet rs, PreparedStatement statement) {
551
 
552
 		logger.logp(Level.FINEST, CLASSNAME, "cleanupConnection", "Entering",  new Object[] {conn, rs==null ? "<null>" : rs, statement==null ? "<null>" : statement});
553
 
554
 		if (statement != null) {
555
 			try {
556
 				statement.close();
557
 			} catch (SQLException e) {
558
 				throw new PersistenceException(e);
559
 			}
560
 		}
561
 
562
 		if (rs != null) {
563
 			try {
564
 				rs.close();
565
 			} catch (SQLException e) {
566
 				throw new PersistenceException(e);
567
 			}
568
 		}
569
 
570
 		if (conn != null) {
571
 			try {
572
 				conn.close();
573
 			} catch (SQLException e) {
574
 				throw new PersistenceException(e);
575
 			} finally {
576
 				try {
577
 					conn.close();
578
 				} catch (SQLException e) {
579
 					throw new PersistenceException(e);
580
 				}
581
 			}
582
 		}
583
 		logger.logp(Level.FINEST, CLASSNAME, "cleanupConnection", "Exiting");
584
 	}

	
 closes connection and statement
 
 
 Parameters:conn - connection object to closestatement - statement object to close
591
 
592
 	private void cleanupConnection(Connection conn, PreparedStatement statement) {
593
 
594
 		logger.logp(Level.FINEST, CLASSNAME, "cleanupConnection", "Entering",  new Object[] {conn, statement});
595
 
596
 		if (statement != null) {
597
 			try {
598
 				statement.close();
599
 			} catch (SQLException e) {
600
 				throw new PersistenceException(e);
601
 			}
602
 		}
603
 
604
 		if (conn != null) {
605
 			try {
606
 				conn.close();
607
 			} catch (SQLException e) {
608
 				throw new PersistenceException(e);
609
 			} finally {
610
 				try {
611
 					conn.close();
612
 				} catch (SQLException e) {
613
 					throw new PersistenceException(e);
614
 				}
615
 			}
616
 		}
617
 		logger.logp(Level.FINEST, CLASSNAME, "cleanupConnection", "Exiting");
618
 	}
619
 
620
 
621
 	@Override
622
 	public int jobOperatorGetJobInstanceCount(String jobName, String appTag) {
623
 		
624
 		Connection conn = null;
625
 		PreparedStatement statement = null;
626
 		ResultSet rs = null;
627
 		int count;
628
 
629
 		try {
630
 			conn = getConnection();
631
 			statement = conn.prepareStatement("select count(jobinstanceid) as jobinstancecount from jobinstancedata where name = ? and apptag = ?");
632
 			statement.setString(1, jobName);
633
 			statement.setString(2, appTag);
634
 			rs = statement.executeQuery();
635
 			rs.next();
636
 			count = rs.getInt("jobinstancecount");
637
 
638
 		} catch (SQLException e) {
639
 			throw new PersistenceException(e);
640
 		}
641
 		finally {
642
 			cleanupConnection(conn, rs, statement);
643
 		}
644
 		return count;
645
 	}
646
 	
647
 	@Override
648
 	public int jobOperatorGetJobInstanceCount(String jobName) {
649
 
650
 		Connection conn = null;
651
 		PreparedStatement statement = null;
652
 		ResultSet rs = null;
653
 		int count;
654
 
655
 		try {
656
 			conn = getConnection();
657
 			statement = conn.prepareStatement(SELECT_JOBINSTANCEDATA_COUNT);
658
 			statement.setString(1, jobName);
659
 			rs = statement.executeQuery();
660
 			rs.next();
661
 			count = rs.getInt("jobinstancecount");
662
 
663
 		} catch (SQLException e) {
664
 			throw new PersistenceException(e);
665
 		}
666
 		finally {
667
 			cleanupConnection(conn, rs, statement);
668
 		}
669
 		return count;
670
 	}
671
 
672
 
673
 	@Override
674
 	public List<Long> jobOperatorGetJobInstanceIds(String jobName, String appTag, int start, int count) {
675
 		Connection conn = null;
676
 		PreparedStatement statement = null;
677
 		ResultSet rs = null;
678
 		List<Long> data = new ArrayList<Long>();
679
 
680
 		try {
681
 			conn = getConnection();
682
 			statement = conn.prepareStatement("select jobinstanceid from jobinstancedata where name = ? and apptag = ? order by jobinstanceid desc");
683
 			statement.setObject(1, jobName);
684
 			statement.setObject(2, appTag);
685
 			rs = statement.executeQuery();
686
 			while (rs.next()) {
687
 				long id = rs.getLong("jobinstanceid");
688
 				data.add(id);
689
 			}
690
 		} catch (SQLException e) {
691
 			throw new PersistenceException(e);
692
 		}
693
 		finally {
694
 			cleanupConnection(conn, rs, statement);
695
 		}
696
 
697
 		if (data.size() > 0){
698
 			try {
699
 				return data.subList(start, start+count);
700
 			}
701
 			catch (IndexOutOfBoundsException oobEx){
702
 				return data.subList(start, data.size());
703
 			}
704
 		}
705
 		else return data;
706
 	}
707
 	
708
 	@Override
709
 	public List<Long> jobOperatorGetJobInstanceIds(String jobName, int start, int count) {
710
 
711
 		Connection conn = null;
712
 		PreparedStatement statement = null;
713
 		ResultSet rs = null;
714
 		List<Long> data = new ArrayList<Long>();
715
 
716
 		try {
717
 			conn = getConnection();
718
 			statement = conn.prepareStatement(SELECT_JOBINSTANCEDATA_IDS);
719
 			statement.setObject(1, jobName);
720
 			rs = statement.executeQuery();
721
 			while (rs.next()) {
722
 				long id = rs.getLong("jobinstanceid");
723
 				data.add(id);
724
 			}
725
 		} catch (SQLException e) {
726
 			throw new PersistenceException(e);
727
 		}
728
 		finally {
729
 			cleanupConnection(conn, rs, statement);
730
 		}
731
 
732
 		if (data.size() > 0){
733
 			try {
734
 				return data.subList(start, start+count);
735
 			}
736
 			catch (IndexOutOfBoundsException oobEx){
737
 				return data.subList(start, data.size());
738
 			}
739
 		}
740
 		else return data;
741
 	}
742
 
743
 	@Override
744
 	public Map<Long, String> jobOperatorGetExternalJobInstanceData() {
745
 
746
 		Connection conn = null;
747
 		PreparedStatement statement = null;
748
 		ResultSet rs = null;
749
 		HashMap<Long, String> data = new HashMap<Long,String>();
750
 
751
 		try {
752
 			conn = getConnection();
753
 
754
 			// Filter out 'subjob' parallel execution entries which start with the special character
755
 			final String filter = "not like '" + PartitionedStepBuilder.JOB_ID_SEPARATOR + "%'";
756
 
757
 			statement = conn.prepareStatement("select distinct jobinstanceid, name from jobinstancedata where name " + filter );
758
 			rs = statement.executeQuery();
759
 			while (rs.next()) {
760
 				long id = rs.getLong("jobinstanceid");
761
 				String name = rs.getString("name");
762
 				data.put(id, name);
763
 			}
764
 		} catch (SQLException e) {
765
 			throw new PersistenceException(e);
766
 		}
767
 		finally {
768
 			cleanupConnection(conn, rs, statement);
769
 		}
770
 
771
 		return data;
772
 	}
773
 
774
 	@Override
775
 	public Timestamp jobOperatorQueryJobExecutionTimestamp(long key, TimestampType timestampType) {
776
 
777
 
778
 
779
 		Connection conn = null;
780
 		PreparedStatement statement = null;
781
 		ResultSet rs = null;
782
 		Timestamp createTimestamp = null;
783
 		Timestamp endTimestamp = null;
784
 		Timestamp updateTimestamp = null;
785
 		Timestamp startTimestamp = null;
786
 
787
 		try {
788
 			conn = getConnection();
789
 			statement = conn.prepareStatement("select createtime, endtime, updatetime, starttime from executioninstancedata where jobexecid = ?");
790
 			statement.setObject(1, key);
791
 			rs = statement.executeQuery();
792
 			while (rs.next()) {
793
 				createTimestamp = rs.getTimestamp(1);
794
 				endTimestamp = rs.getTimestamp(2);
795
 				updateTimestamp = rs.getTimestamp(3);
796
 				startTimestamp = rs.getTimestamp(4);
797
 			}
798
 		} catch (SQLException e) {
799
 			throw new PersistenceException(e);
800
 		}
801
 		finally {
802
 			cleanupConnection(conn, rs, statement);
803
 		}
804
 
805
 		if (timestampType.equals(TimestampType.CREATE)) {
806
 			return createTimestamp;
807
 		} else if (timestampType.equals(TimestampType.END)) {
808
 			return endTimestamp;
809
 		} else if (timestampType.equals(TimestampType.LAST_UPDATED)) {
810
 			return updateTimestamp;
811
 		} else if (timestampType.equals(TimestampType.STARTED)) {
812
 			return startTimestamp;
813
 		} else {
814
 			throw new IllegalArgumentException("Unexpected enum value.");
815
 		}
816
 	}
817
 
818
 	@Override
819
 	public String jobOperatorQueryJobExecutionBatchStatus(long key) {
820
 
821
 		Connection conn = null;
822
 		PreparedStatement statement = null;
823
 		ResultSet rs = null;
824
 		String status = null;
825
 
826
 		try {
827
 			conn = getConnection();
828
 			statement = conn.prepareStatement("select batchstatus from executioninstancedata where jobexecid = ?");
829
 			statement.setLong(1, key);
830
 			rs = statement.executeQuery();
831
 			while (rs.next()) {
832
 				status = rs.getString(1);
833
 			}
834
 		} catch (SQLException e) {
835
 			throw new PersistenceException(e);
836
 		} finally {
837
 			cleanupConnection(conn, rs, statement);
838
 		}
839
 
840
 		return status;
841
 	}
842
 
843
 
844
 	@Override
845
 	public String jobOperatorQueryJobExecutionExitStatus(long key) {
846
 
847
 		Connection conn = null;
848
 		PreparedStatement statement = null;
849
 		ResultSet rs = null;
850
 		String status = null;
851
 
852
 		try {
853
 			conn = getConnection();
854
 			statement = conn.prepareStatement("select exitstatus from executioninstancedata where jobexecid = ?");
855
 			statement.setLong(1, key);
856
 			rs = statement.executeQuery();
857
 			while (rs.next()) {
858
 				status = rs.getString(1);
859
 			}
860
 		} catch (SQLException e) {
861
 			throw new PersistenceException(e);
862
 		} finally {
863
 			cleanupConnection(conn, rs, statement);
864
 		}
865
 
866
 		return status;
867
 	}
868
 
869
 	@Override
870
 	public long jobOperatorQueryJobExecutionJobInstanceId(long executionID) throws NoSuchJobExecutionException {
871
 
872
 		Connection conn = null;
873
 		PreparedStatement statement = null;
874
 		ResultSet rs = null;
875
 		long jobinstanceID = 0;
876
 
877
 		try {
878
 			conn = getConnection();
879
 			statement = conn.prepareStatement("select jobinstanceid from executioninstancedata where jobexecid = ?");
880
 			statement.setLong(1, executionID);
881
 			rs = statement.executeQuery();
882
 			if (rs.next()) {
883
 				jobinstanceID = rs.getLong("jobinstanceid");
884
 			} else {
885
 				String msg = "Did not find job instance associated with executionID =" + executionID;
886
 				logger.fine(msg);
887
 				throw new NoSuchJobExecutionException(msg);
888
 			}
889
 		} catch (SQLException e) {
890
 			throw new PersistenceException(e);
891
 		}
892
 		finally {
893
 			cleanupConnection(conn, rs, statement);
894
 		}
895
 
896
 		return jobinstanceID;
897
 	}
898
 
899
 	@Override
900
 	public Properties getParameters(long executionId) throws NoSuchJobExecutionException {
901
 		Connection conn = null;
902
 		PreparedStatement statement = null;
903
 		ResultSet rs = null;
904
 		Properties props = null;
905
 		ObjectInputStream objectIn = null;
906
 
907
 		try {
908
 			conn = getConnection();
909
 			statement = conn.prepareStatement("select parameters from executioninstancedata where jobexecid = ?"); 
910
 			statement.setLong(1, executionId);
911
 			rs = statement.executeQuery();
912
 			
913
 			if (rs.next()) {
914
 				// get the object based data
915
 				byte[] buf = rs.getBytes("parameters");
916
 				props = (Properties)deserializeObject(buf);
917
 			} else {
918
 				String msg = "Did not find table entry for executionID =" + executionId;
919
 				logger.fine(msg);
920
 				throw new NoSuchJobExecutionException(msg);
921
 			}
922
 
923
 		} catch (SQLException e) {
924
 			throw new PersistenceException(e);
925
 		} catch (IOException e) {
926
 			throw new PersistenceException(e);
927
 		} catch (ClassNotFoundException e) {
928
 			throw new PersistenceException(e);
929
 		} finally {
930
 			if (objectIn != null) {
931
 				try {
932
 					objectIn.close();
933
 				} catch (IOException e) {
934
 					throw new PersistenceException(e);
935
 				}
936
 			}
937
 			cleanupConnection(conn, rs, statement);
938
 		}
939
 
940
 		return props;
941
 
942
 	}
943
 
944
 
945
 	public Map<String, StepExecution> getMostRecentStepExecutionsForJobInstance(long instanceId) {
946
 
947
 		Map<String, StepExecution> data = new HashMap<String, StepExecution>();
948
 
949
 		Connection conn = null;
950
 		PreparedStatement statement = null;
951
 		ResultSet rs = null;
952
 
953
 		long jobexecid = 0;
954
 		long stepexecid = 0;
955
 		String stepname = null;
956
 		String batchstatus = null;
957
 		String exitstatus = null;
958
 		Exception ex = null;
959
 		long readCount =0;
960
 		long writeCount = 0;
961
 		long commitCount = 0;
962
 		long rollbackCount = 0;
963
 		long readSkipCount = 0;
964
 		long processSkipCount = 0;
965
 		long filterCount = 0;
966
 		long writeSkipCount = 0;
967
 		Timestamp startTS = null;
968
 		Timestamp endTS = null;
969
 		StepExecutionImpl stepEx = null;
970
 		ObjectInputStream objectIn = null;
971
 
972
 		try {
973
 			conn = getConnection();
974
 			statement = conn.prepareStatement("select A.* from stepexecutioninstancedata as A inner join executioninstancedata as B on A.jobexecid = B.jobexecid where B.jobinstanceid = ? order by A.stepexecid desc"); 
975
 			statement.setLong(1, instanceId);
976
 			rs = statement.executeQuery();
977
 			while (rs.next()) {
978
 				stepname = rs.getString("stepname");
979
 				if (data.containsKey(stepname)) {
980
 					continue;
981
 				} else {
982
 
983
 					jobexecid = rs.getLong("jobexecid");
984
 					batchstatus = rs.getString("batchstatus");
985
 					exitstatus = rs.getString("exitstatus");
986
 					readCount = rs.getLong("readcount");
987
 					writeCount = rs.getLong("writecount");
988
 					commitCount = rs.getLong("commitcount");
989
 					rollbackCount = rs.getLong("rollbackcount");
990
 					readSkipCount = rs.getLong("readskipcount");
991
 					processSkipCount = rs.getLong("processskipcount");
992
 					filterCount = rs.getLong("filtercount");
993
 					writeSkipCount = rs.getLong("writeSkipCount");
994
 					startTS = rs.getTimestamp("startTime");
995
 					endTS = rs.getTimestamp("endTime");
996
 					// get the object based data
997
 					Serializable persistentData = null;
998
 					byte[] pDataBytes = rs.getBytes("persistentData");
999
 					if (pDataBytes != null) {
1000
						objectIn = new TCCLObjectInputStream(new ByteArrayInputStream(pDataBytes));
1001
						persistentData = (Serializable)objectIn.readObject();
1002
					}
1003
1004
					stepEx = new StepExecutionImpl(jobexecid, stepexecid);
1005
1006
					stepEx.setBatchStatus(BatchStatus.valueOf(batchstatus));
1007
					stepEx.setExitStatus(exitstatus);
1008
					stepEx.setStepName(stepname);
1009
					stepEx.setReadCount(readCount);
1010
					stepEx.setWriteCount(writeCount);
1011
					stepEx.setCommitCount(commitCount);
1012
					stepEx.setRollbackCount(rollbackCount);
1013
					stepEx.setReadSkipCount(readSkipCount);
1014
					stepEx.setProcessSkipCount(processSkipCount);
1015
					stepEx.setFilterCount(filterCount);
1016
					stepEx.setWriteSkipCount(writeSkipCount);
1017
					stepEx.setStartTime(startTS);
1018
					stepEx.setEndTime(endTS);	
1019
					stepEx.setPersistentUserData(persistentData);
1020
1021
					data.put(stepname, stepEx);
1022
				}
1023
			}
1024
		} catch (SQLException e) {
1025
			throw new PersistenceException(e);
1026
		} catch (IOException e) {
1027
			throw new PersistenceException(e);
1028
		} catch (ClassNotFoundException e) {
1029
			throw new PersistenceException(e);
1030
		} finally {
1031
			cleanupConnection(conn, rs, statement);
1032
		}
1033
1034
		return data;
1035
	}
1036
1037
1038
	@Override
1039
	public List<StepExecution> getStepExecutionsForJobExecution(long execid) {
1040
		Connection conn = null;
1041
		PreparedStatement statement = null;
1042
		ResultSet rs = null;
1043
1044
		long jobexecid = 0;
1045
		long stepexecid = 0;
1046
		String stepname = null;
1047
		String batchstatus = null;
1048
		String exitstatus = null;
1049
		Exception ex = null;
1050
		long readCount =0;
1051
		long writeCount = 0;
1052
		long commitCount = 0;
1053
		long rollbackCount = 0;
1054
		long readSkipCount = 0;
1055
		long processSkipCount = 0;
1056
		long filterCount = 0;
1057
		long writeSkipCount = 0;
1058
		Timestamp startTS = null;
1059
		Timestamp endTS = null;
1060
		StepExecutionImpl stepEx = null;
1061
		ObjectInputStream objectIn = null;
1062
1063
		List<StepExecution> data = new ArrayList<StepExecution>();
1064
1065
		try {
1066
			conn = getConnection();
1067
			statement = conn.prepareStatement("select * from stepexecutioninstancedata where jobexecid = ?"); 
1068
			statement.setLong(1, execid);
1069
			rs = statement.executeQuery();
1070
			while (rs.next()) {
1071
				jobexecid = rs.getLong("jobexecid");
1072
				stepexecid = rs.getLong("stepexecid");
1073
				stepname = rs.getString("stepname");
1074
				batchstatus = rs.getString("batchstatus");
1075
				exitstatus = rs.getString("exitstatus");
1076
				readCount = rs.getLong("readcount");
1077
				writeCount = rs.getLong("writecount");
1078
				commitCount = rs.getLong("commitcount");
1079
				rollbackCount = rs.getLong("rollbackcount");
1080
				readSkipCount = rs.getLong("readskipcount");
1081
				processSkipCount = rs.getLong("processskipcount");
1082
				filterCount = rs.getLong("filtercount");
1083
				writeSkipCount = rs.getLong("writeSkipCount");
1084
				startTS = rs.getTimestamp("startTime");
1085
				endTS = rs.getTimestamp("endTime");
1086
				// get the object based data
1087
				Serializable persistentData = null;
1088
				byte[] pDataBytes = rs.getBytes("persistentData");
1089
				if (pDataBytes != null) {
1090
					objectIn = new TCCLObjectInputStream(new ByteArrayInputStream(pDataBytes));
1091
					persistentData = (Serializable)objectIn.readObject();
1092
				}
1093
1094
				stepEx = new StepExecutionImpl(jobexecid, stepexecid);
1095
1096
				stepEx.setBatchStatus(BatchStatus.valueOf(batchstatus));
1097
				stepEx.setExitStatus(exitstatus);
1098
				stepEx.setStepName(stepname);
1099
				stepEx.setReadCount(readCount);
1100
				stepEx.setWriteCount(writeCount);
1101
				stepEx.setCommitCount(commitCount);
1102
				stepEx.setRollbackCount(rollbackCount);
1103
				stepEx.setReadSkipCount(readSkipCount);
1104
				stepEx.setProcessSkipCount(processSkipCount);
1105
				stepEx.setFilterCount(filterCount);
1106
				stepEx.setWriteSkipCount(writeSkipCount);
1107
				stepEx.setStartTime(startTS);
1108
				stepEx.setEndTime(endTS);	
1109
				stepEx.setPersistentUserData(persistentData);
1110
1111
				data.add(stepEx);
1112
			}
1113
		} catch (SQLException e) {
1114
			throw new PersistenceException(e);
1115
		} catch (IOException e) {
1116
			throw new PersistenceException(e);
1117
		} catch (ClassNotFoundException e) {
1118
			throw new PersistenceException(e);
1119
		} finally {
1120
			cleanupConnection(conn, rs, statement);
1121
		}
1122
1123
		return data;
1124
	}
1125
1126
	
1127
    @Override
1128
    public StepExecution getStepExecutionByStepExecutionId(long stepExecId) {
1129
        Connection conn = null;
1130
        PreparedStatement statement = null;
1131
        ResultSet rs = null;
1132