PTAL, thanks. @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);
        }
    }

testAllCompactedOut

Need to add this test to this patch

@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 Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.58%. Comparing base (b25d2c3) to head (6def666).
⚠️ Report is 1319 commits behind head on master.

@@            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