Allow serializing complex objects in JobDataMaps when using Quartz extension

extensions/quartz/deployment/src/main/java/io/quarkus/quartz/deployment/QuartzProcessor.java

@@ -4,6 +4,7 @@
 
 import java.sql.Connection;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -13,25 +14,24 @@
 
 import org.jboss.jandex.DotName;
 import org.quartz.Job;
+import org.quartz.JobDataMap;
 import org.quartz.JobListener;
 import org.quartz.TriggerListener;
 import org.quartz.core.QuartzSchedulerThread;
 import org.quartz.core.SchedulerSignalerImpl;
 import org.quartz.impl.StdSchedulerFactory;
 import org.quartz.impl.jdbcjobstore.AttributeRestoringConnectionInvocationHandler;
-import org.quartz.impl.jdbcjobstore.DB2v8Delegate;
-import org.quartz.impl.jdbcjobstore.HSQLDBDelegate;
 import org.quartz.impl.jdbcjobstore.JobStoreSupport;
-import org.quartz.impl.jdbcjobstore.MSSQLDelegate;
-import org.quartz.impl.jdbcjobstore.PostgreSQLDelegate;
-import org.quartz.impl.jdbcjobstore.StdJDBCDelegate;
 import org.quartz.impl.triggers.AbstractTrigger;
 import org.quartz.impl.triggers.SimpleTriggerImpl;
 import org.quartz.simpl.CascadingClassLoadHelper;
+import org.quartz.simpl.InitThreadContextClassLoadHelper;
 import org.quartz.simpl.SimpleInstanceIdGenerator;
 import org.quartz.simpl.SimpleThreadPool;
 import org.quartz.spi.InstanceIdGenerator;
 import org.quartz.spi.SchedulerPlugin;
+import org.quartz.utils.DirtyFlagMap;
+import org.quartz.utils.StringKeyDirtyFlagMap;
 
 import io.quarkus.agroal.spi.JdbcDataSourceBuildItem;
 import io.quarkus.agroal.spi.JdbcDataSourceSchemaReadyBuildItem;
@@ -58,6 +58,11 @@
 import io.quarkus.quartz.runtime.QuartzRuntimeConfig;
 import io.quarkus.quartz.runtime.QuartzSchedulerImpl;
 import io.quarkus.quartz.runtime.QuartzSupport;
+import io.quarkus.quartz.runtime.jdbc.QuarkusDBv8Delegate;
+import io.quarkus.quartz.runtime.jdbc.QuarkusHSQLDBDelegate;
+import io.quarkus.quartz.runtime.jdbc.QuarkusMSSQLDelegate;
+import io.quarkus.quartz.runtime.jdbc.QuarkusPostgreSQLDelegate;
+import io.quarkus.quartz.runtime.jdbc.QuarkusStdJDBCDelegate;
 import io.quarkus.runtime.configuration.ConfigurationException;
 
 /**
@@ -126,32 +131,40 @@ QuartzJDBCDriverDialectBuildItem driver(List<JdbcDataSourceBuildItem> jdbcDataSo
 
     private String guessDriver(Optional<JdbcDataSourceBuildItem> jdbcDataSource) {
         if (!jdbcDataSource.isPresent()) {
-            return StdJDBCDelegate.class.getName();
+            return QuarkusStdJDBCDelegate.class.getName();
         }
 
         String dataSourceKind = jdbcDataSource.get().getDbKind();
         if (DatabaseKind.isPostgreSQL(dataSourceKind)) {
-            return PostgreSQLDelegate.class.getName();
+            return QuarkusPostgreSQLDelegate.class.getName();
         }
         if (DatabaseKind.isH2(dataSourceKind)) {
-            return HSQLDBDelegate.class.getName();
+            return QuarkusHSQLDBDelegate.class.getName();
         }
         if (DatabaseKind.isMsSQL(dataSourceKind)) {
-            return MSSQLDelegate.class.getName();
+            return QuarkusMSSQLDelegate.class.getName();
         }
         if (DatabaseKind.isDB2(dataSourceKind)) {
-            return DB2v8Delegate.class.getName();
+            return QuarkusDBv8Delegate.class.getName();
         }
 
-        return StdJDBCDelegate.class.getName();
-
+        return QuarkusStdJDBCDelegate.class.getName();
     }
 
     @BuildStep
     List<ReflectiveClassBuildItem> reflectiveClasses(QuartzBuildTimeConfig config,
             QuartzJDBCDriverDialectBuildItem driverDialect) {
         List<ReflectiveClassBuildItem> reflectiveClasses = new ArrayList<>();
 
+        if (config.serializeJobData.orElse(false)) {
+            reflectiveClasses.add(ReflectiveClassBuildItem.builder(
+                    String.class,
+                    JobDataMap.class,
+                    DirtyFlagMap.class,
+                    StringKeyDirtyFlagMap.class,
+                    HashMap.class).serialization(true).build());
+        }
+
         reflectiveClasses
                 .add(ReflectiveClassBuildItem.builder(SimpleThreadPool.class.getName()).methods().build());
         reflectiveClasses.add(ReflectiveClassBuildItem.builder(SimpleInstanceIdGenerator.class.getName()).methods()
@@ -160,7 +173,7 @@ List<ReflectiveClassBuildItem> reflectiveClasses(QuartzBuildTimeConfig config,
                 .build());
         reflectiveClasses.add(ReflectiveClassBuildItem.builder(config.storeType.clazz).methods().fields().build());
         reflectiveClasses
-                .add(ReflectiveClassBuildItem.builder(org.quartz.simpl.InitThreadContextClassLoadHelper.class)
+                .add(ReflectiveClassBuildItem.builder(InitThreadContextClassLoadHelper.class)
                         .build());
 
         if (config.storeType.isDbStore()) {

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzBuildTimeConfig.java

@@ -21,6 +21,8 @@ public class QuartzBuildTimeConfig {
 
     /**
      * The frequency (in milliseconds) at which the scheduler instance checks-in with other instances of the cluster.
+     * <p>
+     * Ignored if using a `ram` store i.e {@link StoreType#RAM}.
      */
     @ConfigItem(defaultValue = "15000")
     public long clusterCheckinInterval;
@@ -43,6 +45,8 @@ public class QuartzBuildTimeConfig {
     /**
      * The name of the datasource to use.
      * <p>
+     * Ignored if using a `ram` store i.e {@link StoreType#RAM}.
+     * <p>
      * Optionally needed when using the `jdbc-tx` or `jdbc-cmt` store types.
      * If not specified, defaults to using the default datasource.
      */
@@ -52,14 +56,16 @@ public class QuartzBuildTimeConfig {
     /**
      * The prefix for quartz job store tables.
      * <p>
-     * Ignored if using a `ram` store.
+     * Ignored if using a `ram` store i.e {@link StoreType#RAM}
      */
     @ConfigItem(defaultValue = "QRTZ_")
     public String tablePrefix;
 
     /**
      * The SQL string that selects a row in the "LOCKS" table and places a lock on the row.
      * <p>
+     * Ignored if using a `ram` store i.e {@link StoreType#RAM}.
+     * <p>
      * If not set, the default value of Quartz applies, for which the "{0}" is replaced during run-time with the
      * `table-prefix`, the "{1}" with the `instance-name`.
      * <p>
@@ -68,6 +74,26 @@ public class QuartzBuildTimeConfig {
     @ConfigItem
     public Optional<String> selectWithLockSql;
 
+    /**
+     * Instructs JDBCJobStore to serialize JobDataMaps in the BLOB column.
+     * <p>
+     * Ignored if using a `ram` store i.e {@link StoreType#RAM}.
+     * <p>
+     * If this is set to `true`, the JDBCJobStore will store the JobDataMaps in their serialize form in the BLOB Column.
+     * This is useful when you want to store complex JobData objects other than String.
+     * This is equivalent of setting `org.quartz.jobStore.useProperties` to `false`.
+     * <b>NOTE: When this option is set to `true`, all the non-String classes used in JobDataMaps have to be registered
+     * for serialization when building a native image</b>
+     * <p>
+     * If this is set to `false` (the default), the values can be stored as name-value pairs rather than storing more complex
+     * objects in their serialized form in the BLOB column.
+     * This can be handy, as you avoid the class versioning issues that can arise from serializing your non-String classes into
+     * a BLOB.
+     * This is equivalent of setting `org.quartz.jobStore.useProperties` to `true`.
+     */
+    @ConfigItem(defaultValue = "false")
+    public Optional<Boolean> serializeJobData;
+
     /**
      * Instance ID generators.
      */

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzSchedulerImpl.java

@@ -528,7 +528,8 @@ private Properties getSchedulerConfigurationProperties(QuartzSupport quartzSuppo
         if (buildTimeConfig.storeType.isDbStore()) {
             String dataSource = buildTimeConfig.dataSourceName.orElse("QUARKUS_QUARTZ_DEFAULT_DATASOURCE");
             QuarkusQuartzConnectionPoolProvider.setDataSourceName(dataSource);
-            props.put(StdSchedulerFactory.PROP_JOB_STORE_PREFIX + ".useProperties", "true");
+            boolean serializeJobData = buildTimeConfig.serializeJobData.orElse(false);
+            props.put(StdSchedulerFactory.PROP_JOB_STORE_USE_PROP, serializeJobData ? "false" : "true");
             props.put(StdSchedulerFactory.PROP_JOB_STORE_PREFIX + ".misfireThreshold",
                     "" + runtimeConfig.misfireThreshold.toMillis());
             props.put(StdSchedulerFactory.PROP_JOB_STORE_PREFIX + ".tablePrefix", buildTimeConfig.tablePrefix);
@@ -1022,5 +1023,4 @@ public Job newJob(TriggerFiredBundle bundle, org.quartz.Scheduler Scheduler) thr
         }
 
     }
-
 }

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/graal/QuartzSubstitutions.java

@@ -1,11 +1,8 @@
 package io.quarkus.quartz.runtime.graal;
 
-import java.io.ByteArrayOutputStream;
 import java.rmi.RemoteException;
-import java.sql.ResultSet;
 
 import org.quartz.core.RemotableQuartzScheduler;
-import org.quartz.impl.jdbcjobstore.StdJDBCDelegate;
 
 import com.oracle.svm.core.annotate.Substitute;
 import com.oracle.svm.core.annotate.TargetClass;
@@ -40,30 +37,5 @@ protected RemotableQuartzScheduler getRemoteScheduler() {
 
 }
 
-@TargetClass(StdJDBCDelegate.class)
-final class Target_org_quartz_impl_jdbc_jobstore_StdJDBCDelegate {
-
-    /**
-     * Activate the usage of {@link java.util.Properties} to avoid Object serialization
-     * which is not supported by GraalVM - see https://github.com/oracle/graal/issues/460
-     *
-     * @return true
-     */
-    @Substitute
-    protected boolean canUseProperties() {
-        return true;
-    }
-
-    @Substitute
-    protected ByteArrayOutputStream serializeObject(Object obj) {
-        throw new IllegalStateException("Object serialization not supported."); // should not reach here
-    }
-
-    @Substitute
-    protected Object getObjectFromBlob(ResultSet rs, String colName) {
-        throw new IllegalStateException("Object serialization not supported."); // should not reach here
-    }
-}
-
 final class QuartzSubstitutions {
 }

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/jdbc/DBDelegateUtils.java

@@ -0,0 +1,26 @@
+package io.quarkus.quartz.runtime.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+
+class DBDelegateUtils {
+    /**
+     * A method to deserialize a marshalled object in an input stream.
+     * This implementation uses {@link QuarkusObjectInputStream} instead of {@link ObjectInputStream} to workaround
+     * a {@link ClassNotFoundException} issue observed in Test & Dev mode when `resolveClass(ObjectStreamClass)` is called.
+     */
+    static Object getObjectFromInput(InputStream binaryInput) throws ClassNotFoundException, IOException {
+        if (binaryInput == null || binaryInput.available() == 0) {
+            return null;
+        }
+        // use an instance of the QuarkusObjectInputStream class instead of the ObjectInputStream when deserializing
+        // to workaround a CNFE in test and dev mode.
+        ObjectInputStream in = new QuarkusObjectInputStream(binaryInput);
+        try {
+            return in.readObject();
+        } finally {
+            in.close();
+        }
+    }
+}

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/jdbc/QuarkusDBv8Delegate.java

@@ -0,0 +1,25 @@
+package io.quarkus.quartz.runtime.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+import java.sql.Blob;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class QuarkusDBv8Delegate extends org.quartz.impl.jdbcjobstore.DB2v8Delegate {
+    /**
+     * See the javadoc in {@link QuarkusObjectInputStream#resolveClass(ObjectStreamClass)} and
+     * {@link DBDelegateUtils#getObjectFromInput(InputStream)}
+     * on why this is needed
+     */
+    @Override
+    protected Object getObjectFromBlob(ResultSet rs, String colName) throws ClassNotFoundException, IOException, SQLException {
+        Blob blobLocator = rs.getBlob(colName);
+        if (blobLocator == null || blobLocator.length() == 0) {
+            return null;
+        }
+        InputStream binaryInput = blobLocator.getBinaryStream();
+        return DBDelegateUtils.getObjectFromInput(binaryInput);
+    }
+}

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/jdbc/QuarkusHSQLDBDelegate.java

@@ -0,0 +1,20 @@
+package io.quarkus.quartz.runtime.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class QuarkusHSQLDBDelegate extends org.quartz.impl.jdbcjobstore.HSQLDBDelegate {
+    /**
+     * See the javadoc in {@link QuarkusObjectInputStream#resolveClass(ObjectStreamClass)} and
+     * {@link DBDelegateUtils#getObjectFromInput(InputStream)}
+     * on why this is needed
+     */
+    @Override
+    protected Object getObjectFromBlob(ResultSet rs, String colName) throws ClassNotFoundException, IOException, SQLException {
+        InputStream binaryInput = rs.getBinaryStream(colName);
+        return DBDelegateUtils.getObjectFromInput(binaryInput);
+    }
+}

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/jdbc/QuarkusMSSQLDelegate.java

@@ -0,0 +1,20 @@
+package io.quarkus.quartz.runtime.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class QuarkusMSSQLDelegate extends org.quartz.impl.jdbcjobstore.HSQLDBDelegate {
+    /**
+     * See the javadoc in {@link QuarkusObjectInputStream#resolveClass(ObjectStreamClass)} and
+     * {@link DBDelegateUtils#getObjectFromInput(InputStream)}
+     * on why this is needed
+     */
+    @Override
+    protected Object getObjectFromBlob(ResultSet rs, String colName) throws ClassNotFoundException, IOException, SQLException {
+        InputStream binaryInput = rs.getBinaryStream(colName);
+        return DBDelegateUtils.getObjectFromInput(binaryInput);
+    }
+}

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/jdbc/QuarkusObjectInputStream.java

@@ -0,0 +1,38 @@
+package io.quarkus.quartz.runtime.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+
+/**
+ * See the javadoc in {@link QuarkusObjectInputStream#resolveClass(ObjectStreamClass)}
+ */
+class QuarkusObjectInputStream extends ObjectInputStream {
+    public QuarkusObjectInputStream(InputStream in) throws IOException {
+        super(in);
+    }
+
+    /**
+     * We override the {@link ObjectInputStream#resolveClass(ObjectStreamClass)} method to workaround a class loading issue
+     * in Test & Dev mode.
+     * This is because, the implementation of this method in ObjectInputStream returns the result of calling
+     * Class.forName(desc.getName(), false, loader)
+     * where loader is the first class loader on the current thread's stack (starting from the currently executing method)
+     * that is neither the platform class loader nor its ancestor; otherwise, loader is the platform class loader.
+     * That classloader happens to the Base Runtime QuarkusClassLoader in Test and Dev mode which was causing
+     * {@link ClassNotFoundException}
+     * when loading user/application classes.
+     *
+     */
+    @Override
+    protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {
+        String name = desc.getName();
+        try {
+            // uses the TCCL to workaround CNFE encountered in test & dev mode
+            return Class.forName(name, false, Thread.currentThread().getContextClassLoader());
+        } catch (ClassNotFoundException ex) {
+            return super.resolveClass(desc);
+        }
+    }
+}

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/jdbc/QuarkusPostgreSQLDelegate.java

@@ -0,0 +1,26 @@
+package io.quarkus.quartz.runtime.jdbc;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class QuarkusPostgreSQLDelegate extends org.quartz.impl.jdbcjobstore.PostgreSQLDelegate {
+    /**
+     * See the javadoc in {@link QuarkusObjectInputStream#resolveClass(ObjectStreamClass)} and
+     * {@link DBDelegateUtils#getObjectFromInput(InputStream)}
+     * on why this is needed
+     */
+    @Override
+    protected Object getObjectFromBlob(ResultSet rs, String colName) throws ClassNotFoundException, IOException, SQLException {
+        byte[] bytes = rs.getBytes(colName);
+        if (bytes == null || bytes.length == 0) {
+            return null;
+        }
+
+        InputStream binaryInput = new ByteArrayInputStream(bytes);
+        return DBDelegateUtils.getObjectFromInput(binaryInput);
+    }
+}