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

pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java

@@ -134,7 +134,11 @@ public static Optional<RawMessage> rebatchMessage(RawMessage msg,
                                                       msg.getMessageIdData().getEntryId(),
                                                       msg.getMessageIdData().getPartition(),
                                                       i);
-                if (!singleMessageMetadata.hasPartitionKey()) {
+                if (singleMessageMetadata.isCompactedOut()) {
+                    // we may read compacted out message from the compacted topic
+                    Commands.serializeSingleMessageInBatchWithPayload(emptyMetadata,
+                            Unpooled.EMPTY_BUFFER, batchBuffer);
+                } else if (!singleMessageMetadata.hasPartitionKey()) {
                     if (retainNullKey) {
                         messagesRetained++;
                         Commands.serializeSingleMessageInBatchWithPayload(singleMessageMetadata,

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java

@@ -36,6 +36,8 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
 import lombok.Cleanup;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.LedgerEntry;
@@ -46,10 +48,15 @@
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.admin.LongRunningProcessStatus;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.ConnectionPool;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.RawMessageImpl;
@@ -177,6 +184,55 @@ public void testCompaction() throws Exception {
         compactAndVerify(topic, expected, true);
     }
 
+    @Test
+    public void testAllCompactedOut() throws Exception {
+        String topicName = "persistent://my-property/use/my-ns/testAllCompactedOut";
+        // set retain null key to true
+        boolean oldRetainNullKey = pulsar.getConfig().isTopicCompactionRetainNullKey();
+        pulsar.getConfig().setTopicCompactionRetainNullKey(true);
+        this.restartBroker();
+
+        @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(() -> {
+            Assert.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(() -> {
+            Assert.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();
+        while (reader.hasMessageAvailable()) {
+            Message<String> message = reader.readNext(3, TimeUnit.SECONDS);
+            Assert.assertNotNull(message);
+        }
+        // set retain null key back to avoid affecting other tests
+        pulsar.getConfig().setTopicCompactionRetainNullKey(oldRetainNullKey);
+    }
+
     @Test
     public void testCompactAddCompact() throws Exception {
         String topic = "persistent://my-property/use/my-ns/my-topic1";