JdbcSearchableJobExecutionDao needs to support LocalDateTime for DB2

* Code now handles the NPE that is thrown by DB2 when an a LocalDateTime field contains null when retrieving from the database.

whooops

Updated based on code review

Added TODO and a log message discussing that this is a fix for a DB2 Driver error

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/batch/JdbcSearchableJobExecutionDao.java

@@ -30,6 +30,8 @@
 
 import javax.sql.DataSource;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.JobExecution;
@@ -38,6 +40,7 @@
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.repository.dao.JdbcJobExecutionDao;
 import org.springframework.batch.item.database.Order;
+import org.springframework.cloud.dataflow.core.database.support.DatabaseType;
 import org.springframework.cloud.dataflow.server.batch.support.SqlPagingQueryProviderFactoryBean;
 import org.springframework.cloud.dataflow.server.converter.StringToDateConverter;
 import org.springframework.cloud.dataflow.server.repository.support.SchemaUtilities;
@@ -49,6 +52,7 @@
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowCallbackHandler;
 import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.support.MetaDataAccessException;
 import org.springframework.jdbc.support.incrementer.AbstractDataFieldMaxValueIncrementer;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -112,6 +116,9 @@ public class JdbcSearchableJobExecutionDao extends JdbcJobExecutionDao implement
 
 	private static final String GET_JOB_EXECUTIONS_BY_TASK_IDS = "SELECT JOB_EXECUTION_ID, TASK_EXECUTION_ID from %TASK_PREFIX%TASK_BATCH WHERE TASK_EXECUTION_ID in (?)";
 
+	private static final Logger logger = LoggerFactory.getLogger(JdbcSearchableJobExecutionDao.class);
+
+
 	private DataflowSqlPagingQueryProvider allExecutionsPagingQueryProvider;
 
 	private DataflowSqlPagingQueryProvider byJobNamePagingQueryProvider;
@@ -137,6 +144,8 @@ public class JdbcSearchableJobExecutionDao extends JdbcJobExecutionDao implement
 
 	private String taskTablePrefix;
 
+	private DatabaseType databaseType;
+
 	public JdbcSearchableJobExecutionDao() {
 		conversionService = new DefaultConversionService();
 		conversionService.addConverter(new StringToDateConverter());
@@ -178,7 +187,7 @@ protected long getNextKey() {
 		byJobInstanceIdWithStepCountPagingQueryProvider = getPagingQueryProvider(FIELDS_WITH_STEP_COUNT, null, JOB_INSTANCE_ID_FILTER);
 		byTaskExecutionIdWithStepCountPagingQueryProvider = getPagingQueryProvider(FIELDS_WITH_STEP_COUNT,
 				getTaskQuery(FROM_CLAUSE_TASK_TASK_BATCH), TASK_EXECUTION_ID_FILTER);
-
+		databaseType = getDatabaseType();
 		super.afterPropertiesSet();
 	}
 
@@ -639,16 +648,44 @@ JobExecution createJobExecutionFromResultSet(ResultSet rs, int rowNum) throws SQ
 		jobExecution = new JobExecution(jobInstance, jobParameters);
 		jobExecution.setId(id);
 
-		jobExecution.setStartTime(rs.getObject(2, LocalDateTime.class));
-		jobExecution.setEndTime(rs.getObject(3, LocalDateTime.class));
+		jobExecution.setStartTime(getLocalDateTime(2, rs));
+		jobExecution.setEndTime(getLocalDateTime(3, rs));
 		jobExecution.setStatus(BatchStatus.valueOf(rs.getString(4)));
 		jobExecution.setExitStatus(new ExitStatus(rs.getString(5), rs.getString(6)));
-		jobExecution.setCreateTime(rs.getObject(7, LocalDateTime.class));
-		jobExecution.setLastUpdated(rs.getObject(8, LocalDateTime.class));
+		jobExecution.setCreateTime(getLocalDateTime(7, rs));
+		jobExecution.setLastUpdated(getLocalDateTime(8, rs));
 		jobExecution.setVersion(rs.getInt(9));
 		return jobExecution;
 	}
 
+	private  LocalDateTime getLocalDateTime(int columnIndex, ResultSet rs) throws SQLException {
+		LocalDateTime result = null;
+		// TODO: When the DB2 driver can support LocalDateTime, remove this if block.
+		if (databaseType == DatabaseType.DB2) {
+			try {
+				result = rs.getObject(columnIndex, LocalDateTime.class);
+			} catch (NullPointerException npe) {
+				if (!npe.getMessage().contains("java.sql.Timestamp.toLocalDateTime()\" because \"<local4>\" is null")) {
+					throw npe;
+				}
+				logger.debug("DB2 threw a NPE because it fails to handle an empty column for a java LocalDateTime .  SCDF returns a null for this column.");
+			}
+		} else {
+			result = rs.getObject(columnIndex, LocalDateTime.class);
+		}
+		return result;
+	}
+
+	private DatabaseType getDatabaseType() throws SQLException {
+		DatabaseType databaseType;
+		try {
+			databaseType = DatabaseType.fromMetaData(dataSource);
+		} catch (MetaDataAccessException e) {
+			throw new IllegalStateException(e);
+		}
+		return databaseType;
+	}
+
 	private final class JobExecutionRowMapper implements RowMapper<JobExecution> {
 
 		private JobInstance jobInstance;

spring-cloud-dataflow-server/src/test/java/org/springframework/cloud/dataflow/server/db/migration/DB2_11_5_SmokeTest.java

@@ -28,14 +28,6 @@
  * @author Corneil du Plessis
  * @author Chris Bono
  */
-//TODO: Boot3x - DB2 Driver has a bug.
-//java.lang.NullPointerException: Cannot invoke "java.sql.Timestamp.toLocalDateTime()" because "<local4>" is null
-//at com.ibm.db2.jcc.am.ResultSet.getObject(ResultSet.java:2020)
-//at com.ibm.db2.jcc.am.ResultSet.getObject(ResultSet.java:2045)
-//at com.zaxxer.hikari.pool.HikariProxyResultSet.getObject(HikariProxyResultSet.java)
-//at org.springframework.cloud.task.repository.dao.JdbcTaskExecutionDao$TaskExecutionRowMapper.mapRow(JdbcTaskExecutionDao.java:621)
-//TODO: Boot3x followup
-@Disabled("TODO: Boot3x DB2 Driver and LocalDateTime has a bug when the row has is null in the column")
 @EnabledIfEnvironmentVariable(named = "ENABLE_DB2", matches = "true", disabledReason = "Container is too big")
 @Tag("DB2")
 public class DB2_11_5_SmokeTest extends AbstractSmokeTest implements DB2_11_5_ContainerSupport {