apache / apache/pulsar

[Bug] Dead Letter Queue not working in custom sink due to ClassNotFoundException

Open
#21,138 3 comments 0 reactions 0 assignees View on GitHub
Stale type/bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.