QuarkusClassLoader bannedResources and Apache Beam

[cfinkelstein]

Hello, I was wondering if I can use the Quarkus Console Runner to execute my Apache Beam pipeline. Apache Beam serializes an Element before it will be handed over to the next pipeline step. But it does not work because the QuarkusClassLoader will be used here which bans all the project classes.

Can someone explain what the concept of banned resources is and why it does not work here?

[cfinkelstein]

Here is my example:

@QuarkusMain
public class Main implements QuarkusApplication {

    @Inject
    LastFetchReader lastFetchReader;

    @Override
    public int run(String... args) throws Exception {
        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline pipeline = Pipeline.create(options);
        pipeline.apply(lastFetchReader.read());
        pipeline.run();
        return 0;
    }
}

@ApplicationScoped
public class LastFetchReader implements Serializable {

    private class LastFetchRowMapper implements JdbcIO.RowMapper<LastFetchDate> {
        @Override
        public LastFetchDate mapRow(@UnknownKeyFor @NonNull @Initialized ResultSet resultSet) throws @UnknownKeyFor@NonNull@Initialized Exception {
            var fetchDate = new LastFetchDate();
            fetchDate.setVendor(resultSet.getString(1));
            fetchDate.setValue(resultSet.getTimestamp(2).toInstant().atZone(ZoneId.of("UTC")));
            return fetchDate;
        }
    }

    public Read<LastFetchDate> read() {

        JdbcIO.DataSourceConfiguration dataSourceConfiguration = JdbcIO.DataSourceConfiguration.create(
                "com.mysql.cj.jdbc.Driver",
                "jdbc:mysql://<...>"
        ).withUsername("...").withPassword("...");


        return JdbcIO.<LastFetchDate>read()
                .withDataSourceConfiguration(dataSourceConfiguration)
                .withQuery("SELECT vendor, fetch_time FROM ...")
                .withRowMapper(new LastFetchRowMapper());
    }
}

2023-03-17 10:37:51,319 ERROR [io.qua.run.Application] (Quarkus Main Thread) Failed to start application (with profile [dev]): java.lang.ClassNotFoundException: com.test.webinar.steps.models.LastFetchDate
	at io.quarkus.bootstrap.classloading.QuarkusClassLoader.loadClass(QuarkusClassLoader.java:489)
	at io.quarkus.bootstrap.classloading.QuarkusClassLoader.loadClass(QuarkusClassLoader.java:466)
	at java.base/java.lang.Class.forName0(Native Method)
	at java.base/java.lang.Class.forName(Class.java:495)
	at java.base/java.lang.Class.forName(Class.java:474)
	at java.base/java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:802)
	at java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2059)
	at java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1925)
	at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2248)
	at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1760)
	at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:538)
	at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:496)
	at org.apache.beam.sdk.coders.SerializableCoder.decode(SerializableCoder.java:200)
	at org.apache.beam.sdk.coders.SerializableCoder.decode(SerializableCoder.java:57)
	at org.apache.beam.sdk.coders.Coder.decode(Coder.java:159)
	at org.apache.beam.sdk.util.CoderUtils.decodeFromSafeStream(CoderUtils.java:142)
	at org.apache.beam.sdk.util.CoderUtils.decodeFromByteArray(CoderUtils.java:102)
	at org.apache.beam.sdk.util.CoderUtils.decodeFromByteArray(CoderUtils.java:96)
	at org.apache.beam.sdk.util.CoderUtils.clone(CoderUtils.java:168)
	at org.apache.beam.sdk.util.MutationDetectors$CodedValueMutationDetector.<init>(MutationDetectors.java:118)
	at org.apache.beam.sdk.util.MutationDetectors.forValueWithCoder(MutationDetectors.java:49)
	at org.apache.beam.runners.direct.ImmutabilityCheckingBundleFactory$ImmutabilityEnforcingBundle.add(ImmutabilityCheckingBundleFactory.java:115)
	at org.apache.beam.runners.direct.ParDoEvaluator$BundleOutputManager.output(ParDoEvaluator.java:305)
	at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:275)
	at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.access$900(SimpleDoFnRunner.java:85)
	at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:423)
	at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:411)
	at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn.processElement(JdbcIO.java:1498)
	at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn$DoFnInvoker.invokeProcessElement(Unknown Source)
	at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:211)
	at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188)
	at org.apache.beam.repackaged.direct_java.runners.core.SimplePushbackSideInputDoFnRunner.processElementInReadyWindows(SimplePushbackSideInputDoFnRunner.java:79)
	at org.apache.beam.runners.direct.ParDoEvaluator.processElement(ParDoEvaluator.java:244)
	at org.apache.beam.runners.direct.DoFnLifecycleManagerRemovingTransformEvaluator.processElement(DoFnLifecycleManagerRemovingTransformEvaluator.java:54)
	at org.apache.beam.runners.direct.DirectTransformExecutor.processElements(DirectTransformExecutor.java:165)
	at org.apache.beam.runners.direct.DirectTransformExecutor.run(DirectTransformExecutor.java:129)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:577)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1589)

See: https://github.com/quarkusio/quarkus/blob/main/independent-projects/bootstrap/core/src/main/java/io/quarkus/bootstrap/classloading/QuarkusClassLoader.java - Line 489

[gsmet]

Is it in dev mode or with regular jar?

I think a small self contained reproducer would help if either of you could prepare something.

/cc @geoand for your awareness