[pulsar-client] Creating Consumer with SubscriptionMode.NonDurable fails with NPE

[lhotari]

Describe the bug

Creating a consumer with SubscriptionMode.NonDurable fails with NPE .

To Reproduce

Any consumer with NonDurable subscription mode fails with NPE.

For example:

Consumer<String> consumer = client.newConsumer(Schema.STRING)
                .topic("topic")
                .subscriptionName("subscription")
                .subscriptionType(SubscriptionType.Exclusive)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscriptionMode(SubscriptionMode.NonDurable)
                .subscribe();
consumer.receive();

logs:

19:31:09.602 [pulsar-client-io-1-1] WARN  o.a.p.client.impl.ConnectionHandler - [ca8aeb67-bc66-4053-92ac-c139fb1f9e80] [82823470-a6b8-4fa1-af56-83f25f750c26] Error connecting to broker: java.lang.NullPointerException
19:31:09.617 [pulsar-client-io-1-1] INFO  o.a.pulsar.client.impl.ConsumerImpl - [ca8aeb67-bc66-4053-92ac-c139fb1f9e80] Consumer creation failed for consumer 3

with a debugger I was able to capture the stacktrace of the NPE, it's this:

java.lang.NullPointerException
	at org.apache.pulsar.client.impl.ConsumerImpl.connectionOpened(ConsumerImpl.java:586)
	at org.apache.pulsar.client.impl.ConnectionHandler.lambda$grabCnx$0(ConnectionHandler.java:68)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire$$$capture(CompletableFuture.java:714)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
	at org.apache.pulsar.client.impl.BinaryProtoLookupService.lambda$null$2(BinaryProtoLookupService.java:127)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire$$$capture(CompletableFuture.java:714)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
	at org.apache.pulsar.client.impl.ClientCnx.handleLookupResponse(ClientCnx.java:480)
	at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:132)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:475)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:834)

with the debugger I can see that startMessageId is null here:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 586 in d243808

builder.setLedgerId(startMessageId.getLedgerId());

Expected behavior

Pulsar client would support non durable subscription mode for consumers.

Additional context
Pulsar client 2.5.2 (on Java 11), Pulsar broker 2.5.2 (in Docker)

[lhotari]

I wonder if PR #7355 will fix this issue. I'll test it once 2.6.1 is available in maven central.

[lhotari]

It looks like there is no NPE with 2.6.1 , however the issue now is that .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) has no impact when using .subscriptionMode(SubscriptionMode.NonDurable) . Existing message in the topic don't get consumed. New messages get consumed which didn't work at all in 2.5.2 so some improvement has been made. Does it make sense to create a separate issue about .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) not working with .subscriptionMode(SubscriptionMode.NonDurable)?

[codelipenghui]

It looks like there is no NPE with 2.6.1 , however the issue now is that .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) has no impact when using .subscriptionMode(SubscriptionMode.NonDurable) . Existing message in the topic don't get consumed. New messages get consumed which didn't work at all in 2.5.2 so some improvement has been made. Does it make sense to create a separate issue about .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) not working with .subscriptionMode(SubscriptionMode.NonDurable)?

Interesting, could you please share more details of your test? such as did you set up the data retention? The Non-durable consumer is equaled to Reader, If the Non-durable consumer can't work, I think the reader API also can't work.

[tisonkun]

Verified cannot reproduce on the master branch, including the following Earlist issue.

Close.