Add support for custom JdbcCustomization via Spring Boot (#357)

* fixes #354 different behaviour of jTDS MSSQL driver

* Make JdbcCustomization configurable for spring boot starter as well.

* Add missing serialVersionUID for Serializable classes

* Deleting MssqlJtdsJdbcCustomization. Needs to be supplied from the dependee.

---------

Co-authored-by: Gustav Karlsson <kagkarlsson@gmail.com>

db-scheduler-boot-starter/src/main/java/com/github/kagkarlsson/scheduler/boot/autoconfigure/DbSchedulerAutoConfiguration.java

@@ -19,6 +19,7 @@
 import com.github.kagkarlsson.scheduler.Scheduler;
 import com.github.kagkarlsson.scheduler.SchedulerBuilder;
 import com.github.kagkarlsson.scheduler.SchedulerName;
+import com.github.kagkarlsson.scheduler.jdbc.AutodetectJdbcCustomization;
 import com.github.kagkarlsson.scheduler.serializer.Serializer;
 import com.github.kagkarlsson.scheduler.boot.config.DbSchedulerCustomizer;
 import com.github.kagkarlsson.scheduler.boot.config.DbSchedulerProperties;
@@ -29,6 +30,7 @@
 import com.github.kagkarlsson.scheduler.stats.StatsRegistry;
 import com.github.kagkarlsson.scheduler.task.OnStartup;
 import com.github.kagkarlsson.scheduler.task.Task;
+
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.ObjectInput;
@@ -39,6 +41,7 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import javax.sql.DataSource;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
@@ -143,6 +146,11 @@ public Scheduler scheduler(DbSchedulerCustomizer customizer, StatsRegistry regis
         // Use custom serializer if provided. Otherwise use devtools friendly serializer.
         builder.serializer(customizer.serializer().orElse(SPRING_JAVA_SERIALIZER));
 
+        // Use custom JdbcCustomizer if provided.
+        builder.jdbcCustomization(
+            customizer.jdbcCustomization()
+                .orElse(new AutodetectJdbcCustomization(transactionalDataSource)));
+
         if (config.isImmediateExecutionEnabled()) {
             builder.enableImmediateExecution();
         }

db-scheduler-boot-starter/src/main/java/com/github/kagkarlsson/scheduler/boot/config/DbSchedulerCustomizer.java

@@ -16,7 +16,9 @@
 package com.github.kagkarlsson.scheduler.boot.config;
 
 import com.github.kagkarlsson.scheduler.SchedulerName;
+import com.github.kagkarlsson.scheduler.jdbc.JdbcCustomization;
 import com.github.kagkarlsson.scheduler.serializer.Serializer;
+
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 
@@ -45,4 +47,11 @@ default Optional<Serializer> serializer() {
     default Optional<ExecutorService> executorService() {
         return Optional.empty();
     }
+
+    /**
+     * Provide a custom JdbcCustomization.
+     */
+    default Optional<JdbcCustomization> jdbcCustomization() {
+        return Optional.empty();
+    }
 }

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/jdbc/AutodetectJdbcCustomization.java

@@ -67,6 +67,16 @@ public Instant getInstant(ResultSet rs, String columnName) throws SQLException {
         return jdbcCustomization.getInstant(rs, columnName);
     }
 
+    @Override
+    public void setTaskData(PreparedStatement p, int index, byte[] value) throws SQLException {
+        jdbcCustomization.setTaskData(p, index, value);
+    }
+
+    @Override
+    public byte[] getTaskData(ResultSet rs, String columnName) throws SQLException {
+        return jdbcCustomization.getTaskData(rs, columnName);
+    }
+
     @Override
     public boolean supportsExplicitQueryLimitPart() {
         return jdbcCustomization.supportsExplicitQueryLimitPart();

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/jdbc/DefaultJdbcCustomization.java

@@ -37,6 +37,16 @@ public Instant getInstant(ResultSet rs, String columnName) throws SQLException {
         return Optional.ofNullable(rs.getTimestamp(columnName)).map(Timestamp::toInstant).orElse(null);
     }
 
+    @Override
+    public void setTaskData(PreparedStatement p, int index, byte[] value) throws SQLException {
+        p.setObject(index, value);
+    }
+
+    @Override
+    public byte[] getTaskData(ResultSet rs, String columnName) throws SQLException {
+        return rs.getBytes(columnName);
+    }
+
     @Override
     public boolean supportsExplicitQueryLimitPart() {
         return false;

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/jdbc/JdbcCustomization.java

@@ -29,6 +29,10 @@ public interface JdbcCustomization {
     void setInstant(PreparedStatement p, int index, Instant value) throws SQLException;
     Instant getInstant(ResultSet rs, String columnName) throws SQLException;
 
+    void setTaskData(PreparedStatement p, int index, byte[] value) throws SQLException;
+    byte[] getTaskData(ResultSet rs, String columnName) throws SQLException;
+
+
     boolean supportsExplicitQueryLimitPart();
     String getQueryLimitPart(int limit);
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/jdbc/JdbcTaskRepository.java

@@ -103,7 +103,7 @@ public boolean createIfNotExists(SchedulableInstance instance) {
                     (PreparedStatement p) -> {
                         p.setString(1, taskInstance.getTaskName());
                         p.setString(2, taskInstance.getId());
-                        p.setObject(3, serializer.serialize(taskInstance.getData()));
+                        jdbcCustomization.setTaskData(p, 3, serializer.serialize(taskInstance.getData()));
                         jdbcCustomization.setInstant(p, 4, instance.getNextExecutionTime(clock.now()));
                         p.setBoolean(5, false);
                         p.setLong(6, 1L);
@@ -470,7 +470,7 @@ public Void map(ResultSet rs) throws SQLException {
                 }
 
                 String instanceId = rs.getString("task_instance");
-                byte[] data = rs.getBytes("task_data");
+                byte[] data = jdbcCustomization.getTaskData(rs,"task_data");
 
                 Instant executionTime = jdbcCustomization.getInstant(rs, "execution_time");
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/jdbc/MssqlJdbcCustomization.java

@@ -27,7 +27,7 @@
 import java.util.Optional;
 import java.util.TimeZone;
 
-public class MssqlJdbcCustomization implements JdbcCustomization {
+public class MssqlJdbcCustomization extends DefaultJdbcCustomization {
 
     @Override
     public String getName() {
@@ -39,28 +39,4 @@ public void setInstant(PreparedStatement p, int index, Instant value) throws SQL
         p.setTimestamp(index, value != null ? Timestamp.from(value) : null, Calendar.getInstance(TimeZone.getTimeZone("UTC")));
     }
 
-    @Override
-    public Instant getInstant(ResultSet rs, String columnName) throws SQLException {
-        return Optional.ofNullable(rs.getTimestamp(columnName)).map(Timestamp::toInstant).orElse(null);
-    }
-
-    @Override
-    public boolean supportsExplicitQueryLimitPart() {
-        return false;
-    }
-
-    @Override
-    public String getQueryLimitPart(int limit) {
-        return "";
-    }
-
-    @Override
-    public boolean supportsLockAndFetch() {
-        return false;
-    }
-
-    @Override
-    public List<Execution> lockAndFetch(JdbcTaskRepositoryContext ctx, Instant now, int limit) {
-        throw new UnsupportedOperationException("lockAndFetch not supported for " + this.getClass().getName());
-    }
 }

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/helper/PlainScheduleAndData.java

@@ -21,6 +21,7 @@
 import java.util.Objects;
 
 public class PlainScheduleAndData implements ScheduleAndData, Serializable {
+    private static final long serialVersionUID = 1L;
     private final Schedule schedule;
     private final Object data;
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/schedule/CronSchedule.java

@@ -37,8 +37,11 @@
  */
 public class CronSchedule implements Schedule, Serializable {
 
+    private static final long serialVersionUID = 1L;
+
     private static final String DISABLED = "-";
     private static final Logger LOG = LoggerFactory.getLogger(CronSchedule.class);
+
     private final String pattern;
     private final ZoneId zoneId;
     private transient ExecutionTime cronExecutionTime; // lazily initialized

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/schedule/Daily.java

@@ -30,6 +30,8 @@
 
 public class Daily implements Schedule, Serializable {
 
+    private static final long serialVersionUID = 1L;
+
     private final List<LocalTime> times;
     private final ZoneId zone;
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/schedule/FixedDelay.java

@@ -24,6 +24,8 @@
 
 public class FixedDelay implements Schedule, Serializable {
 
+    private static final long serialVersionUID = 1L;
+
     private final Duration duration;
 
     private FixedDelay() { // For serializers

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/schedule/PersistentCronSchedule.java

@@ -20,6 +20,7 @@
 import java.time.ZoneId;
 
 public class PersistentCronSchedule implements ScheduleAndData {
+    private static final long serialVersionUID = 1L;
     private final String cronPattern;
     private final String zoneId;
     private final Object data;

examples/spring-boot-example/src/main/java/com/github/kagkarlsson/examples/boot/config/BasicExamplesConfiguration.java

@@ -15,8 +15,6 @@
  */
 package com.github.kagkarlsson.examples.boot.config;
 
-import static com.github.kagkarlsson.scheduler.task.schedule.Schedules.fixedDelay;
-
 import com.github.kagkarlsson.examples.boot.CounterService;
 import com.github.kagkarlsson.examples.boot.ExampleContext;
 import com.github.kagkarlsson.scheduler.task.Task;
@@ -31,6 +29,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import static com.github.kagkarlsson.scheduler.task.schedule.Schedules.fixedDelay;
+
 @Configuration
 public class BasicExamplesConfiguration {