[pulsar-client] Fix send to deadLetterTopic not working when reach maxRedeliverCount

[Shawyeok]

Motivation

If a message reached maxRedeliverCount, it will send to deadLetterTopic, since 2.8.0, this mechanism is broken, it was introduced in #9970

You can reproduce with code below:

class Scratch {

    private static final Logger LOG = LoggerFactory.getLogger(Scratch.class);

    public static void main(String[] args) throws Exception {
        PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl(args[0])
                .build();
        DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
                .maxRedeliverCount(0)
                .build();
        Consumer<byte[]> consumer = pulsarClient.newConsumer()
                .topic(args[1])
                .subscriptionName("consumeTest")
                .subscriptionType(SubscriptionType.Shared)
                .deadLetterPolicy(deadLetterPolicy)
                .ackTimeout(10, TimeUnit.SECONDS)
                .subscribe();
        int i = 1;
        Message<byte[]> message;
        while ((message = consumer.receive()) != null) {
            MessageId messageId = message.getMessageId();
            LOG.info("Receive a message {}: {}", messageId, new String(message.getData()));
            if (i-- > 0) {
                LOG.info("Skip a message {}", messageId);
                continue;
            }
            consumer.acknowledge(messageId);
        }
        new CountDownLatch(1).await();
    }
}

It will log exception below:

Dead letter producer exception with topic: {{topic}}
java.util.concurrent.CompletionException: java.lang.NullPointerException
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
	at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
	at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:659)
	at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:632)
	at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:443)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.NullPointerException
	at java.util.Objects.requireNonNull(Objects.java:203)
	at org.apache.pulsar.client.impl.schema.AbstractSchema.atSchemaVersion(AbstractSchema.java:81)
	at org.apache.pulsar.client.impl.MessageImpl.getReaderSchema(MessageImpl.java:398)
	at org.apache.pulsar.client.impl.ConsumerImpl.lambda$processPossibleToDLQ$38(ConsumerImpl.java:1692)
	at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656)
	... 6 more

    // MessageImpl#getReaderSchema
    public Optional<Schema<?>> getReaderSchema() {
        ensureSchemaIsLoaded();
        if (schema == null) {
            return Optional.empty();
        }
        if (schema instanceof AutoConsumeSchema) {
            byte[] schemaVersion = getSchemaVersion();
            return Optional.of(((AutoConsumeSchema) schema)
                    .atSchemaVersion(schemaVersion));
        } else if (schema instanceof AbstractSchema) {
            byte[] schemaVersion = getSchemaVersion();    // schemaVersion may be null,  e.g. BYTES schema
            return Optional.of(((AbstractSchema<?>) schema)
                    .atSchemaVersion(schemaVersion));           // if schemaVersion is null, a NPE will throw
        } else {
            return Optional.of(schema);
        }
    }

Modifications

Make AbstractSchema#atSchemaVersion throw NPE only if supportSchemaVersioning is true and schemaVersion is null.

Verifying this change

This change added tests and can be verified as follows:

• Added a unit test for MessageImpl#getReaderSchema with a message which has BYTES schema

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below or label this PR directly (if you have committer privilege).

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

Bugfix only

• doc

  (If this PR contains doc changes)

[Shawyeok]

Could you please take a look about this? @hangc0276 @congbobo184

[Technoboy-]

LGTM

[michaeljmarshall]

@Shawyeok - Why was this line necessary? It seems like it's not doing anything.

[Shawyeok]

Indeed, should I open a PR to remove this line?