[Bug] Dead Letter Queue not working in custom sink due to ClassNotFoundException
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Version
Pulsar 2.8.4
### Minimal reproduce step
We are experiencing a ClassNotFoundException when a Pulsar custom sink tries to produce to a DLQ topic once processing of a message has failed (and so the retries). The class is inside `pulsar-client-original` JAR file present in Pulsar docker image and also in the packaged NAR file of our sink.
The error is the following:
```
17:19:24.006 [ForkJoinPool.commonPool-worker-3] ERROR org.apache.pulsar.client.impl.ConsumerImpl - Dead letter producer exception with topic: persistent://dbus/test/dummy-objects-sub-test-rtsink-postgresql-DLQ
java.util.concurrent.CompletionException: java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassNotFoundException: org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314) ~[?:?]
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319) ~[?:?]
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718) ~[?:?]
at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:479) ~[?:?]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290) ~[?:?]
at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020) ~[?:?]
at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656) ~[?:?]
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594) ~[?:?]
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183) ~[?:?]
Caused by: java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassNotFoundException: org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema
at org.apache.pulsar.client.internal.ReflectionUtils.catchExceptions(ReflectionUtils.java:45) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.DefaultImplementation.newAutoProduceSchema(DefaultImplementation.java:295) ~[java-instance.jar:?]
at org.apache.pulsar.client.api.Schema.AUTO_PRODUCE_BYTES(Schema.java:426) ~[java-instance.jar:?]
at org.apache.pulsar.client.impl.ConsumerImpl.lambda$processPossibleToDLQ$39(ConsumerImpl.java:1706) ~[?:?]
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714) ~[?:?]
... 6 more
Caused by: java.lang.RuntimeException: java.lang.ClassNotFoundException: org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema
at org.apache.pulsar.client.internal.ReflectionUtils.newClassInstance(ReflectionUtils.java:62) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.ReflectionUtils.getConstructor(ReflectionUtils.java:68) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.DefaultImplementation.lambda$newAutoProduceSchema$34(DefaultImplementation.java:296) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.ReflectionUtils.catchExceptions(ReflectionUtils.java:34) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.DefaultImplementation.newAutoProduceSchema(DefaultImplementation.java:295) ~[java-instance.jar:?]
at org.apache.pulsar.client.api.Schema.AUTO_PRODUCE_BYTES(Schema.java:426) ~[java-instance.jar:?]
at org.apache.pulsar.client.impl.ConsumerImpl.lambda$processPossibleToDLQ$39(ConsumerImpl.java:1706) ~[?:?]
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714) ~[?:?]
... 6 more
Caused by: java.lang.ClassNotFoundException: org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema
at jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) ~[?:?]
at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) ~[?:?]
at java.lang.ClassLoader.loadClass(ClassLoader.java:522) ~[?:?]
at java.lang.Class.forName0(Native Method) ~[?:?]
at java.lang.Class.forName(Class.java:398) ~[?:?]
at org.apache.pulsar.client.internal.ReflectionUtils.newClassInstance(ReflectionUtils.java:59) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.ReflectionUtils.getConstructor(ReflectionUtils.java:68) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.DefaultImplementation.lambda$newAutoProduceSchema$34(DefaultImplementation.java:296) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.ReflectionUtils.catchExceptions(ReflectionUtils.java:34) ~[java-instance.jar:?]
at org.apache.pulsar.client.internal.DefaultImplementation.newAutoProduceSchema(DefaultImplementation.java:295) ~[java-instance.jar:?]
at org.apache.pulsar.client.api.Schema.AUTO_PRODUCE_BYTES(Schema.java:426) ~[java-instance.jar:?]
at org.apache.pulsar.client.impl.ConsumerImpl.lambda$processPossibleToDLQ$39(ConsumerImpl.java:1706) ~[?:?]
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714) ~[?:?]
... 6 more
```
It can be easily reproduced by trying to instantiate the missing class (`AutoProduceBytesSchema`) in the `open()` method of the sink using same classloader with following piece of code.
```
@Override
public void open(Map config, SinkContext sinkContext) throws Exception {
// This fails
Class clazz = ClassLoader.getSystemClassLoader().loadClass("org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema");
logger.info("AutoProduceBytesSchema loaded from: {}", clazz.getProtectionDomain().getCodeSource().getLocation());
// Rest of the code
}
```
The error now is the following:
```
18:16:47.173 [dbus/test/rtsink-postgresql-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - Sink open produced uncaught exception:
java.lang.ClassNotFoundException: org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema
at jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) ~[?:?]
at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) ~[?:?]
at java.lang.ClassLoader.loadClass(ClassLoader.java:522) ~[?:?]
at com.kapschtraffic.oneatms.databus.rtsink.RealtimeSink.open(RealtimeSink.java:54) ~[EJP2V0kUmSDLN4DXH0z-jw/:?]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.setupOutput(JavaInstanceRunnable.java:814) ~[?:?]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.setup(JavaInstanceRunnable.java:223) ~[?:?]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:249) ~[?:?]
at java.lang.Thread.run(Thread.java:829) ~[?:?]
18:16:47.248 [dbus/test/rtsink-postgresql-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [dbus/test/rtsink-postgresql:0] Uncaught exception in Java Instance
java.lang.ClassNotFoundException: org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema
at jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) ~[?:?]
at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) ~[?:?]
at java.lang.ClassLoader.loadClass(ClassLoader.java:522) ~[?:?]
at com.kapschtraffic.oneatms.databus.rtsink.RealtimeSink.open(RealtimeSink.java:54) ~[?:?]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.setupOutput(JavaInstanceRunnable.java:814) ~[org.apache.pulsar-pulsar-functions-instance-2.8.4.jar:2.8.4]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.setup(JavaInstanceRunnable.java:223) ~[org.apache.pulsar-pulsar-functions-instance-2.8.4.jar:2.8.4]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:249) ~[org.apache.pulsar-pulsar-functions-instance-2.8.4.jar:2.8.4]
at java.lang.Thread.run(Thread.java:829) ~[?:?]
```
### What did you expect to see?
No exception thrown and message published to the DLQ topic
### What did you see instead?
A ClassNotFoundException
### Anything else?
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
Contributor guide
Research direction
Start by tracing ConsumerImpl.processPossibleToDLQ and Schema.AUTO_PRODUCE_BYTES through DefaultImplementation and ReflectionUtils, then compare the class-loading context used by a custom sink's open method. Check the reported ClassNotFoundException with the reproducible sink setup. Done means a custom sink can publish failed messages to its DLQ without the missing-class exception.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100