[fix] [broker] fix write all compacted out entry into compacted topic

[thetumbled]

Fixes #21916

Motivation

The method reader.hasMessageAvailable() return true, but reader.readNext can't return any messages because the messages in compacted topic are all compacted out.

Modifications

Check if the message without partition key is a valid message.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

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

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: thetumbled#33

[thetumbled]

PTAL, thanks. @coderzc

[codelipenghui]

Is it possible to add a test to reproduce this issue?

[coderzc]

@thetumbled @codelipenghui The following test can reproduce the problem, and this PR can fix it.

    @Test
    public void testAllCompactedOut() throws Exception {
        String topicName = "persistent://my-property/use/my-ns/testAllCompactedOut";

        @Cleanup
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .enableBatching(true).topic(topicName).batchingMaxMessages(3).create();

        producer.newMessage().key("K1").value("V1").sendAsync();
        producer.newMessage().key("K2").value("V2").sendAsync();
        producer.newMessage().key("K2").value(null).sendAsync();
        producer.flush();

        admin.topics().triggerCompaction(topicName);

        Awaitility.await().untilAsserted(() -> {
            assertEquals(admin.topics().compactionStatus(topicName).status,
                    LongRunningProcessStatus.Status.SUCCESS);
        });

        producer.newMessage().key("K1").value(null).sendAsync();
        producer.flush();

        admin.topics().triggerCompaction(topicName);

        Awaitility.await().untilAsserted(() -> {
            assertEquals(admin.topics().compactionStatus(topicName).status,
                    LongRunningProcessStatus.Status.SUCCESS);
        });

        @Cleanup
        Reader<String> reader = pulsarClient.newReader(Schema.STRING)
                .subscriptionName("reader-test")
                .topic(topicName)
                .readCompacted(true)
                .startMessageId(MessageId.earliest)
                .create();
        Assert.assertEquals(reader.getLastMessageIds().get(0), MessageIdImpl.earliest);
        while (reader.hasMessageAvailable()) {
            Message<String> message = reader.readNext(3, TimeUnit.SECONDS);
            Assert.assertNotNull(message);
        }
    }

[Technoboy-]

testAllCompactedOut

Need to add this test to this patch

[thetumbled]

@thetumbled @codelipenghui The following test can reproduce the problem, and this PR can fix it.

    @Test
    public void testAllCompactedOut() throws Exception {
        String topicName = "persistent://my-property/use/my-ns/testAllCompactedOut";

        @Cleanup
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .enableBatching(true).topic(topicName).batchingMaxMessages(3).create();

        producer.newMessage().key("K1").value("V1").sendAsync();
        producer.newMessage().key("K2").value("V2").sendAsync();
        producer.newMessage().key("K2").value(null).sendAsync();
        producer.flush();

        admin.topics().triggerCompaction(topicName);

        Awaitility.await().untilAsserted(() -> {
            assertEquals(admin.topics().compactionStatus(topicName).status,
                    LongRunningProcessStatus.Status.SUCCESS);
        });

        producer.newMessage().key("K1").value(null).sendAsync();
        producer.flush();

        admin.topics().triggerCompaction(topicName);

        Awaitility.await().untilAsserted(() -> {
            assertEquals(admin.topics().compactionStatus(topicName).status,
                    LongRunningProcessStatus.Status.SUCCESS);
        });

        @Cleanup
        Reader<String> reader = pulsarClient.newReader(Schema.STRING)
                .subscriptionName("reader-test")
                .topic(topicName)
                .readCompacted(true)
                .startMessageId(MessageId.earliest)
                .create();
        Assert.assertEquals(reader.getLastMessageIds().get(0), MessageIdImpl.earliest);
        while (reader.hasMessageAvailable()) {
            Message<String> message = reader.readNext(3, TimeUnit.SECONDS);
            Assert.assertNotNull(message);
        }
    }

Great job, but there is something wrong with the test code. We need to set the config topicCompactionRetainNullKey to be true, so that the empty can be written into the compacted topic and the exception is throw out.
I have pushed up the test code, PTAL, thanks! @coderzc

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (b25d2c3) 73.57% compared to head (6def666) 73.58%.
Report is 5 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #21917   +/-   ##
=========================================
  Coverage     73.57%   73.58%           
- Complexity    32375    32395   +20     
=========================================
  Files          1861     1861           
  Lines        138568   138591   +23     
  Branches      15185    15185           
=========================================
+ Hits         101958   101978   +20     
- Misses        28704    28730   +26     
+ Partials       7906     7883   -23     

Flag	Coverage Δ
inttests	24.16% <0.00%> (+0.02%)	⬆️
systests	23.61% <0.00%> (-0.01%)	⬇️
unittests	72.88% <100.00%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...g/apache/pulsar/client/impl/RawBatchConverter.java	93.90% <100.00%> (+0.15%)	⬆️

... and 85 files with indirect coverage changes