apache / apache/pulsar

Support for tombstones (null value in messages) does not work

Open
#7,407 11 comments 1 reaction 0 assignees View on GitHub
help wanted lifecycle/stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**Describe the bug**

The solution provided by #7139 to the BUG #4803 *does not work*.

* When trying to read a message with `null` value a `NullPointerException` is thrown in other part of the code.
```
msg = consumer.receive(timeoutMillis, TimeUnit.MILLISECONDS));

java.lang.NullPointerException
at org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl.updateNumMsgsReceived(ConsumerStatsRecorderImpl.java:169)
at org.apache.pulsar.client.impl.ConsumerImpl.messageProcessed(ConsumerImpl.java:1423)
at org.apache.pulsar.client.impl.ConsumerImpl.internalReceive(ConsumerImpl.java:431)
at org.apache.pulsar.client.impl.ConsumerBase.receive(ConsumerBase.java:175)
...
```
* It should not be required to explicitly indicate a null value to producer (only-key values should work just just fine). Exception thrown when working with implicit null value messages is EOFException in this case (the same before this a solution was provided).

**To Reproduce**
I have created a test project to reproduce these issues (`null` values implicitly and explictly set with both schema and schemaless consumer): [pulsar-tombstone-test](https://github.com/fmiguelez/pulsar-tombstone-test)

Read README.md to reproduce it.

**Expected behavior**
Tombstones (`null` values in mesages with our without schema but with key) should be supported whether you indicate an schema or not and whether you explicitly indicate a `null` value or not (implicit `null` value). All tests should pass in the example project.

```
Producer schemalessProducer = client.newProducer(Schema.BYTES).topic(TOPIC).messageRoutingMode(MessageRoutingMode.SinglePartition).create();

Consumer schemalessConsumer = client.newConsumer(Schema.BYTES).topic(TOPIC).subscriptionName("test")
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();

// Implicit tombstone without schema
schemalessProducer.key("1").send();

// Explicit tombstone (the one supposedly to work)
schemalessProducer.key("2").value(null).send();

Message implicitTombstone = schemalessReceiver.receive(2, TimeUnit.SECONDS).getValue());
Message explictitTombstone = schemalessReceiver.receive(2, TimeUnit.SECONDS).getValue())

System.out.println(String.format("Implicit tombstone: {key=%s, value=%s}", implicitTombstone.getKey(), implicitTombstone.getValue()));
System.out.println(String.format("Explicit tombstone: {key=%s, value=%s}", implicitTombstone.getKey(), implicitTombstone.getValue()));
```

```
Producer schemaProducer = client.newProducer(Schema.AVRO(DummyObject.class)).topic(TOPIC).messageRoutingMode(MessageRoutingMode.SinglePartition).create();

Consumer schemaConsumer = client.newConsumer(Schema.AVRO(DummyObject.class)).topic(TOPIC).subscriptionName("test")
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();

// Implicit tombstone without schema
schemaProducer.key("1").send();

// Explicit tombstone (the one supposedly to work)
schemaProducer.key("2").value(null).send();

Message implicitTombstone = schemalessReceiver.receive(2, TimeUnit.SECONDS).getValue());
Message explictitTombstone = schemalessReceiver.receive(2, TimeUnit.SECONDS).getValue())

System.out.println(String.format("Implicit tombstone: {key=%s, value=%s}", implicitTombstone.getKey(), implicitTombstone.getValue()));
System.out.println(String.format("Explicit tombstone: {key=%s, value=%s}", implicitTombstone.getKey(), implicitTombstone.getValue()));
```

**Screenshots**

**Desktop (please complete the following information):**
- Windows 10 with Docker Deskto to run Pulsar containers

**Additional context**

Contributor guide

Open the contributing guide

Research direction

Start with the README.md in the linked pulsar-tombstone-test reproduction and trace the reported failure through ConsumerStatsRecorderImpl.java, ConsumerImpl.java, and ConsumerBase.java. Reproduce implicit and explicit tombstones with schema and schemaless consumers; done means keyed null messages are supported in all four cases and the reproduction tests pass.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
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.