You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2023/01/20 01:58:19 UTC

[GitHub] [pulsar] heesung-sn opened a new pull request, #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

heesung-sn opened a new pull request, #19045:
URL: https://github.com/apache/pulsar/pull/19045

   <!--
   ### Contribution Checklist
     
     - PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://docs.google.com/document/d/1d8Pw6ZbWk-_pCKdOmdvx9rnhPiyuxwq60_TrD68d7BA/edit#heading=h.trs9rsex3xom)*. 
   
     - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
     
     - Each pull request should address only one issue, not mix up code from multiple issues.
     
     - Each commit in the pull request has a meaningful commit message
   
     - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
   -->
   
   
   <!-- or this PR is one task of an issue -->
   
   Master Issue: https://github.com/apache/pulsar/issues/16691
   
   ### Motivation
   
   <!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
   This PR adds ServiceUnitStateCompactionStrategy.
   
   ### Modifications
   
   For the PIP-192, this PR adds the `ServiceUnitStateCompactionStrategy` and its unit tests.
   
   Also, this PR updates the related classes to enable ServiceUnitStateCompactionStrategy:
   - Added the `strategicCompactor` member in PulsarService.
   - Enabled `ServiceUnitStateCompactionStrategy` for the `tableview` member in ServiceUnitStateChannelImpl.
   - Added the `strategicCompactionMap` member in PersistentTopic
   
   
   <!-- Describe the modifications you've done. -->
   
   ### Verifying this change
   
   - [x] Make sure that the change passes the CI checks.
   
   This change added tests and can be verified as follows:
   
     - *Added unit tests for the new classes.
   
   ### 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
   - [ ] Anything that affects deployment
   
   ### Documentation
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ] `doc` <!-- Your PR contains doc changes. Please attach the local preview screenshots (run `sh start.sh` at `pulsar/site2/website`) to your PR description, or else your PR might not get merged. -->
   - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
   - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-complete` <!-- Docs have been already added -->
   
   We will have separate PRs to update the Doc later.
   
   ### Matching PR in forked repository
   
   PR in forked repository: https://github.com/heesung-sn/pulsar/pull/18
   
   <!--
   After opening this PR, the build in apache/pulsar will fail and instructions will
   be provided for opening a PR in the PR author's forked repository.
   
   apache/pulsar pull requests should be first tested in your own fork since the 
   apache/pulsar CI based on GitHub Actions has constrained resources and quota.
   GitHub Actions provides separate quota for pull requests that are executed in 
   a forked repository.
   
   The tests will be run in the forked repository until all PR review comments have
   been handled, the tests pass and the PR is approved by a reviewer.
   -->
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by GitBox <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1058534838


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategy.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.extensions.channel;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Released;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+
+public class ServiceUnitStateCompactionStrategy implements TopicCompactionStrategy<ServiceUnitStateData> {
+
+    private final Schema<ServiceUnitStateData> schema;
+
+    private boolean checkBrokers = true;
+
+    public ServiceUnitStateCompactionStrategy() {
+        schema = JSONSchema.of(ServiceUnitStateData.class);
+    }
+
+    @Override
+    public Schema<ServiceUnitStateData> getSchema() {
+        return schema;
+    }
+
+    public void checkBrokers(boolean check) {

Review Comment:
   Updated.



##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java:
##########
@@ -0,0 +1,927 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.isValidTransition;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.buffer.ByteBuf;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.Cleanup;
+import lombok.SneakyThrows;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateCompactionStrategy;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.persistent.SystemTopic;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+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.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.protocol.Markers;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-compaction")
+public class ServiceUnitStateCompactionTest extends MockedPulsarServiceBaseTest {
+    private ScheduledExecutorService compactionScheduler;
+    private BookKeeper bk;
+    private JSONSchema<ServiceUnitStateData> schema;
+    private ServiceUnitStateCompactionStrategy strategy;
+
+    private ServiceUnitState testState0 = Free;
+    private ServiceUnitState testState1 = Free;
+    private ServiceUnitState testState2 = Free;
+    private ServiceUnitState testState3 = Free;
+    private ServiceUnitState testState4 = Free;
+
+    private static Random RANDOM = new Random();
+
+
+    private ServiceUnitStateData testValue(ServiceUnitState state, String broker) {
+        if (state == Free) {
+            return null;
+        }
+        return new ServiceUnitStateData(state, broker);
+    }
+
+    private ServiceUnitStateData testValue0(String broker) {
+        ServiceUnitState to = nextValidState(testState0);
+        testState0 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue1(String broker) {
+        ServiceUnitState to = nextValidState(testState1);
+        testState1 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue2(String broker) {
+        ServiceUnitState to = nextValidState(testState2);
+        testState2 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue3(String broker) {
+        ServiceUnitState to = nextValidState(testState3);
+        testState3 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue4(String broker) {
+        ServiceUnitState to = nextValidState(testState4);
+        testState4 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitState nextValidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> to != Free && to != Splitting && isValidTransition(from, to))
+                .collect(Collectors.toList());
+        var state=  candidates.get(RANDOM.nextInt(candidates.size()));
+        System.out.println("test state:" + state);
+        return state;
+    }
+
+    private ServiceUnitState nextInvalidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> !isValidTransition(from, to))
+                .collect(Collectors.toList());
+        if (candidates.size() == 0) {
+            return null;
+        }
+        return candidates.get(RANDOM.nextInt(candidates.size()));
+    }
+
+    private List<ServiceUnitState> nextStatesToNull(ServiceUnitState from) {
+        if (from == null) {
+            return List.of();
+        }
+        return switch (from) {
+            case Assigned -> List.of(Owned);
+            case Owned -> List.of();
+            case Splitting -> List.of();
+            default -> List.of();
+        };
+    }
+
+    @BeforeMethod
+    @Override
+    public void setup() throws Exception {
+        super.internalSetup();
+
+        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+        admin.tenants().createTenant("my-property",
+                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use")));
+        admin.namespaces().createNamespace("my-property/use/my-ns");
+
+        compactionScheduler = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());
+        bk = pulsar.getBookKeeperClientFactory().create(this.conf, null, null, Optional.empty(), null);
+        schema = JSONSchema.of(ServiceUnitStateData.class);
+        strategy = new ServiceUnitStateCompactionStrategy();
+        strategy.checkBrokers(false);
+
+    }
+
+
+    @AfterMethod(alwaysRun = true)
+    @Override
+    public void cleanup() throws Exception {
+        super.internalCleanup();
+
+        if (compactionScheduler != null) {
+            compactionScheduler.shutdownNow();
+        }
+    }
+
+
+    public record TestData(
+            String topic,
+            Map<String, ServiceUnitStateData> expected,
+            List<Pair<String, ServiceUnitStateData>> all) {
+
+    }
+    TestData generateTestData() throws PulsarAdminException, PulsarClientException {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+        final int numMessages = 20;
+        final int maxKeys = 5;
+
+        // Configure retention to ensue data is retained for reader
+        admin.namespaces().setRetention("my-property/use/my-ns", new RetentionPolicies(-1, -1));
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+
+        Map<String, ServiceUnitStateData> expected = new HashMap<>();
+        List<Pair<String, ServiceUnitStateData>> all = new ArrayList<>();
+        Random r = new Random(0);
+
+        pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub1")
+                .readCompacted(true)
+                .subscribe().close();
+
+        for (int j = 0; j < numMessages; j++) {
+            int keyIndex = r.nextInt(maxKeys);
+            String key = "key" + keyIndex;
+            ServiceUnitStateData expectedData = expected.get(key);
+            ServiceUnitState state = expectedData == null ? nextValidState(Assigned) :
+                    r.nextInt(2) == 0 ? nextInvalidState(expectedData.state()) :
+                            nextValidState(expectedData.state());
+            ServiceUnitStateData value = new ServiceUnitStateData(state, key + ":" + j);
+            producer.newMessage().key(key).value(value).send();
+            ServiceUnitStateData prev = expected.get(key);
+            if (!strategy.shouldKeepLeft(prev, value)) {
+                expected.put(key, value);
+            }
+            all.add(Pair.of(key, value));
+        }
+        return new TestData(topic, expected, all);
+    }
+
+    @Test
+    public void testCompaction() throws Exception {
+        TestData testData = generateTestData();
+        var topic = testData.topic;
+        var expected = testData.expected;
+        var all = testData.all;
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic, false);
+        // Compacted topic ledger should have same number of entry equals to number of unique key.
+        //Assert.assertEquals(internalStats.compactedLedger.entries, expected.size());
+        Assert.assertTrue(internalStats.compactedLedger.ledgerId > -1);
+        Assert.assertFalse(internalStats.compactedLedger.offloaded);
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+                Assert.assertEquals(expected.remove(m.getKey()), m.getValue());
+                if (expected.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(expected.isEmpty());
+        }
+
+        // can get full backlog if read compacted disabled
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(false).subscribe()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+                Pair<String, ServiceUnitStateData> expectedMessage = all.remove(0);
+                Assert.assertEquals(expectedMessage.getLeft(), m.getKey());
+                Assert.assertEquals(expectedMessage.getRight(), m.getValue());
+                if (all.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(all.isEmpty());
+        }
+    }
+
+    @Test
+    public void testCompactionWithReader() throws Exception {
+        TestData testData = generateTestData();
+        var topic = testData.topic;
+        var expected = testData.expected;
+        var all = testData.all;
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Reader<ServiceUnitStateData> reader = pulsarClient.newReader(schema).topic(topic).readCompacted(true)
+                .startMessageId(MessageId.earliest).create()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = reader.readNext(2, TimeUnit.SECONDS);
+                Assert.assertEquals(expected.remove(m.getKey()), m.getValue());
+                if (expected.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(expected.isEmpty());
+        }
+
+        // can get full backlog if read compacted disabled
+        try (Reader<ServiceUnitStateData> reader = pulsarClient.newReader(schema).topic(topic).readCompacted(false)
+                .startMessageId(MessageId.earliest).create()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = reader.readNext(2, TimeUnit.SECONDS);
+                Pair<String, ServiceUnitStateData> expectedMessage = all.remove(0);
+                Assert.assertEquals(expectedMessage.getLeft(), m.getKey());
+                Assert.assertEquals(expectedMessage.getRight(), m.getValue());
+                if (all.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(all.isEmpty());
+        }
+    }
+
+
+    @Test
+    public void testCompactionWithTableview() throws Exception {
+        var tv = pulsar.getClient().newTableViewBuilder(schema)
+                .topic("persistent://my-property/use/my-ns/my-topic1")
+                .loadConf(Map.of(
+                        "topicCompactionStrategyClassName",
+                        ServiceUnitStateCompactionStrategy.class.getName()))
+                .create();
+
+        ((ServiceUnitStateCompactionStrategy)
+                FieldUtils.readDeclaredField(tv, "compactionStrategy", true))
+                .checkBrokers(false);
+        TestData testData = generateTestData();
+        var topic = testData.topic;
+        var expected = testData.expected;
+        var expectedCopy = new HashMap<>(expected);
+
+        Awaitility.await()
+                .pollInterval(200, TimeUnit.MILLISECONDS)
+                .atMost(10, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertEquals(expectedCopy.size(), tv.size()));
+
+        for(var etr : tv.entrySet()){
+            Assert.assertEquals(expectedCopy.remove(etr.getKey()), etr.getValue());
+            if (expectedCopy.isEmpty()) {
+                break;
+            }
+        }
+
+        Assert.assertTrue(expectedCopy.isEmpty());
+        tv.close();;
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        var tableview = pulsar.getClient().newTableViewBuilder(schema)
+                .topic(topic)
+                .loadConf(Map.of(
+                        "topicCompactionStrategyClassName",
+                        ServiceUnitStateCompactionStrategy.class.getName()))
+                .create();
+        for(var etr : tableview.entrySet()){
+            Assert.assertEquals(expected.remove(etr.getKey()), etr.getValue());
+            if (expected.isEmpty()) {
+                break;
+            }
+        }
+        Assert.assertTrue(expected.isEmpty());
+        tableview.close();
+
+    }
+
+
+    @Test
+    public void testReadCompactedBeforeCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content0");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content1");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+    }
+
+    @Test
+    public void testReadEntriesAfterCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        producer.newMessage().key("key0").value(testValue0("content3")).send();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content3");
+        }
+    }
+
+    @Test
+    public void testSeekEarliestAfterCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            consumer.seek(MessageId.earliest);
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(false).subscribe()) {
+            consumer.seek(MessageId.earliest);
+
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content0");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content1");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+    }
+
+    @Test
+    public void testBrokerRestartAfterCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+
+        stopBroker();
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            consumer.receive();
+            Assert.fail("Shouldn't have been able to receive anything");
+        } catch (PulsarClientException e) {
+            // correct behaviour
+        }
+        startBroker();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+    }
+
+    @Test
+    public void testCompactEmptyTopic() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content0");
+        }
+    }
+
+    @Test
+    public void testWholeBatchCompactedOut() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        // subscribe before sending anything, so that we get all messages
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe().close();
+
+        try (Producer<ServiceUnitStateData> producerNormal = pulsarClient.newProducer(schema).topic(topic)
+                .enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+             Producer<ServiceUnitStateData> producerBatch = pulsarClient.newProducer(schema).topic(topic)
+                     .maxPendingMessages(3)
+                     .enableBatching(true)
+                     .batchingMaxMessages(3)
+                     .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+                     .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                     .create()) {
+            producerBatch.newMessage().key("key1").value(testValue1("my-message-1")).sendAsync();
+            producerBatch.newMessage().key("key1").value(testValue1( "my-message-2")).sendAsync();
+            producerBatch.newMessage().key("key1").value(testValue1("my-message-3")).sendAsync();
+            producerNormal.newMessage().key("key1").value(testValue1( "my-message-4")).send();
+        }
+
+        // compact the topic
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic)
+                .subscriptionName("sub1").readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> message = consumer.receive();
+            Assert.assertEquals(message.getKey(), "key1");
+            Assert.assertEquals(new String(message.getValue().broker()), "my-message-4");
+        }
+    }
+
+    public void testCompactionWithLastDeletedKey() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema).topic(topic).enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition).create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("1").value(testValue(Owned, "1")).send();
+        producer.newMessage().key("2").value(testValue(Owned, "3")).send();
+        producer.newMessage().key("3").value(testValue(Owned, "5")).send();
+        producer.newMessage().key("1").value(null).send();
+        producer.newMessage().key("2").value(null).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        Set<String> expected = Sets.newHashSet("3");
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+            assertTrue(expected.remove(m.getKey()));
+        }
+    }
+
+    @Test(timeOut = 20000)
+    public void testEmptyCompactionLedger() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema).topic(topic).enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition).create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("1").value(testValue(Owned, "1")).send();
+        producer.newMessage().key("2").value(testValue(Owned, "3")).send();
+        producer.newMessage().key("1").value(null).send();
+        producer.newMessage().key("2").value(null).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(m);
+        }
+    }
+
+    @Test(timeOut = 20000)
+    public void testAllEmptyCompactionLedger() throws Exception {
+        final String topic =
+                "persistent://my-property/use/my-ns/testAllEmptyCompactionLedger" + UUID.randomUUID().toString();
+
+        final int messages = 10;
+
+        // 1.create producer and publish message to the topic.
+        ProducerBuilder<ServiceUnitStateData> builder = pulsarClient.newProducer(schema).topic(topic);
+        builder.batchingMaxMessages(messages / 5);
+
+        Producer<ServiceUnitStateData> producer = builder.create();
+
+        List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key("1").value(null).sendAsync());
+        }
+
+        FutureUtil.waitForAll(futures).get();
+
+        // 2.compact the topic.
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(m);
+        }
+    }
+
+    @Test(timeOut = 20000)
+    public void testCompactMultipleTimesWithoutEmptyMessage()
+            throws PulsarClientException, ExecutionException, InterruptedException {
+        final String topic =
+                "persistent://my-property/use/my-ns/testCompactMultipleTimesWithoutEmptyMessage" + UUID.randomUUID()
+                        .toString();
+
+        final int messages = 10;
+        final String key = "1";
+
+        // 1.create producer and publish message to the topic.
+        ProducerBuilder<ServiceUnitStateData> builder = pulsarClient.newProducer(schema).topic(topic);
+        builder.enableBatching(true);
+
+
+        Producer<ServiceUnitStateData> producer = builder.create();
+
+        List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + ""))).sendAsync());
+        }
+
+        FutureUtil.waitForAll(futures).get();
+
+        // 2.compact the topic.
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // 3. Send more ten messages
+        futures.clear();
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + 10 + ""))).sendAsync());
+        }
+        FutureUtil.waitForAll(futures).get();
+
+        // 4.compact again.
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+            Message<ServiceUnitStateData> m1 = consumer.receive();
+            assertNotNull(m1);
+            assertEquals(m1.getKey(), key);
+            assertEquals(m1.getValue().broker(), "19");
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+    }
+
+    @Test(timeOut = 200000)
+    public void testReadUnCompacted()
+            throws PulsarClientException, ExecutionException, InterruptedException {
+        final String topic = "persistent://my-property/use/my-ns/testReadUnCompacted" + UUID.randomUUID().toString();
+
+        final int messages = 10;
+        final String key = "1";
+
+        // 1.create producer and publish message to the topic.
+        ProducerBuilder<ServiceUnitStateData> builder = pulsarClient.newProducer(schema).topic(topic);
+        builder.batchingMaxMessages(messages / 5);
+
+        Producer<ServiceUnitStateData> producer = builder.create();
+
+        List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + ""))).sendAsync());
+        }
+
+        FutureUtil.waitForAll(futures).get();
+
+        // 2.compact the topic.
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // 3. Send more ten messages
+        futures.clear();
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + 10 + ""))).sendAsync());
+        }
+        FutureUtil.waitForAll(futures).get();
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub1")
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe()) {
+            for (int i = 0; i < 11; i++) {
+                Message<ServiceUnitStateData> received = consumer.receive();
+                assertNotNull(received);
+                assertEquals(received.getKey(), key);
+                assertEquals(received.getValue().broker(), i + 9 + "");
+                consumer.acknowledge(received);
+            }
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+
+        // 4.Send empty message to delete the key-value in the compacted topic.
+        for (ServiceUnitState state : nextStatesToNull(testState0)) {
+            producer.newMessage().key(key).value(new ServiceUnitStateData(state, "xx")).send();
+        }
+        producer.newMessage().key(key).value(null).send();
+
+        // 5.compact the topic.
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub2")
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe()) {
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + 20 + ""))).sendAsync());
+        }
+        FutureUtil.waitForAll(futures).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub3")
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe()) {
+            for (int i = 0; i < 10; i++) {
+                Message<ServiceUnitStateData> received = consumer.receive();
+                assertNotNull(received);
+                assertEquals(received.getKey(), key);
+                assertEquals(received.getValue().broker(), i + 20 + "");
+                consumer.acknowledge(received);
+            }
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+    }
+
+    @SneakyThrows
+    @Test
+    public void testHealthCheckTopicNotCompacted() {
+        NamespaceName heartbeatNamespaceV1 =
+                NamespaceService.getHeartbeatNamespace(pulsar.getAdvertisedAddress(), pulsar.getConfiguration());
+        String topicV1 = "persistent://" + heartbeatNamespaceV1.toString() + "/healthcheck";
+        NamespaceName heartbeatNamespaceV2 =
+                NamespaceService.getHeartbeatNamespaceV2(pulsar.getAdvertisedAddress(), pulsar.getConfiguration());
+        String topicV2 = heartbeatNamespaceV2.toString() + "/healthcheck";
+        Producer<ServiceUnitStateData> producer1 = pulsarClient.newProducer(schema).topic(topicV1).create();
+        Producer<ServiceUnitStateData> producer2 = pulsarClient.newProducer(schema).topic(topicV2).create();
+        Optional<Topic> topicReferenceV1 = pulsar.getBrokerService().getTopic(topicV1, false).join();
+        Optional<Topic> topicReferenceV2 = pulsar.getBrokerService().getTopic(topicV2, false).join();
+        assertFalse(((SystemTopic) topicReferenceV1.get()).isCompactionEnabled());
+        assertFalse(((SystemTopic) topicReferenceV2.get()).isCompactionEnabled());
+        producer1.close();
+        producer2.close();
+    }
+
+    @Test(timeOut = 60000)
+    public void testCompactionWithMarker() throws Exception {
+        String namespace = "my-property/use/my-ns";
+        final TopicName dest = TopicName.get(
+                BrokerTestUtil.newUniqueName("persistent://" + namespace + "/testWriteMarker"));
+        admin.topics().createNonPartitionedTopic(dest.toString());
+        @Cleanup
+        Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(dest.toString())
+                .subscriptionName("test-compaction-sub")
+                .subscriptionType(SubscriptionType.Exclusive)
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
+                .subscribe();

Review Comment:
   In fact, this case is an unnecessary copy from CompactionTest.  Removed it.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] codelipenghui commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.

codelipenghui commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1091329738


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1563,6 +1571,8 @@ public void checkCompaction() {
                 }
 
                 if (backlogEstimate > compactionThreshold) {
+                    log.info("topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering compaction",

Review Comment:
   Suppose you have many topics, like 50k topics per broker. By default, we will have 50k logs per minute. And we already have logs after the compaction task started.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] Technoboy- commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.

Technoboy- commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1089589647


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -619,14 +618,13 @@ private void doCleanup(String broker) {
                     || StringUtils.equals(broker, stateData.sourceBroker())) {
                 log.info("Cleaning ownership serviceUnit:{}, stateData:{}.", serviceUnit, stateData);
                 tombstoneAsync(serviceUnit).whenComplete((__, e) -> {
-                    if (e == null) {
-                        serviceUnitTombstoneCnt.incrementAndGet();
-                    } else {
+                    if (e != null) {
                         log.error("Failed cleaning the ownership serviceUnit:{}, stateData:{}.",
                                 serviceUnit, stateData);
                         serviceUnitTombstoneErrorCnt.incrementAndGet();
                     }
                 });
+                serviceUnitTombstoneCnt.incrementAndGet();

Review Comment:
   Could you help explain why do we need modify here ?
   we update `serviceUnitTombstoneErrorCnt` and `serviceUnitTombstoneCnt` in the async method. these values could both be 0.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] Technoboy- closed pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by GitBox <gi...@apache.org>.

Technoboy- closed pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy
URL: https://github.com/apache/pulsar/pull/19045


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] eolivelli commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "eolivelli (via GitHub)" <gi...@apache.org>.

eolivelli commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1083715317


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategy.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.extensions.channel;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Released;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+
+public class ServiceUnitStateCompactionStrategy implements TopicCompactionStrategy<ServiceUnitStateData> {
+
+    private final Schema<ServiceUnitStateData> schema;
+
+    private boolean checkBrokers = true;
+
+    public ServiceUnitStateCompactionStrategy() {
+        schema = JSONSchema.of(ServiceUnitStateData.class);

Review Comment:
   Please use SchemaJSON()



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1090438614


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -202,6 +205,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(COMPACTION_NEVER_RUN);
     private final CompactedTopic compactedTopic;
 
+    // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users.
+    private static Map<String, TopicCompactionStrategy> strategicCompactionMap = Map.of(
+            ServiceUnitStateChannelImpl.TOPIC,
+            new ServiceUnitStateCompactionStrategy());

Review Comment:
   I am trying to avoid introducing a new member var in PersistentTopic as its memory unnecessarily multiplies as many as topics.
   
   We can remove this hashmap and move each topic compaction strategy config to topic policy once Pulsar exposes this strategic compaction to users.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#issuecomment-1407982677

   @Technoboy- ping 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] codelipenghui merged pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.

codelipenghui merged PR #19045:
URL: https://github.com/apache/pulsar/pull/19045


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] codelipenghui commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.

codelipenghui commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1091330402


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -202,6 +205,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(COMPACTION_NEVER_RUN);
     private final CompactedTopic compactedTopic;
 
+    // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users.
+    private static Map<String, TopicCompactionStrategy> strategicCompactionMap = Map.of(
+            ServiceUnitStateChannelImpl.TOPIC,
+            new ServiceUnitStateCompactionStrategy());

Review Comment:
   Yes, maybe we can just add it to the topic policy now, the HierarchyTopicPolicies. Just do not expose any admin APIs.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#issuecomment-1399827591

   @Technoboy- Hi, can you take a look again?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by GitBox <gi...@apache.org>.

Demogorgon314 commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1058305195


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategy.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.extensions.channel;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Released;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+
+public class ServiceUnitStateCompactionStrategy implements TopicCompactionStrategy<ServiceUnitStateData> {
+
+    private final Schema<ServiceUnitStateData> schema;
+
+    private boolean checkBrokers = true;
+
+    public ServiceUnitStateCompactionStrategy() {
+        schema = JSONSchema.of(ServiceUnitStateData.class);
+    }
+
+    @Override
+    public Schema<ServiceUnitStateData> getSchema() {
+        return schema;
+    }
+
+    public void checkBrokers(boolean check) {

Review Comment:
   Add `@VisibleForTesting` annotation.



##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java:
##########
@@ -0,0 +1,927 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.isValidTransition;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.buffer.ByteBuf;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.Cleanup;
+import lombok.SneakyThrows;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateCompactionStrategy;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.persistent.SystemTopic;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+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.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.protocol.Markers;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-compaction")
+public class ServiceUnitStateCompactionTest extends MockedPulsarServiceBaseTest {
+    private ScheduledExecutorService compactionScheduler;
+    private BookKeeper bk;
+    private JSONSchema<ServiceUnitStateData> schema;
+    private ServiceUnitStateCompactionStrategy strategy;
+
+    private ServiceUnitState testState0 = Free;
+    private ServiceUnitState testState1 = Free;
+    private ServiceUnitState testState2 = Free;
+    private ServiceUnitState testState3 = Free;
+    private ServiceUnitState testState4 = Free;
+
+    private static Random RANDOM = new Random();
+
+
+    private ServiceUnitStateData testValue(ServiceUnitState state, String broker) {
+        if (state == Free) {
+            return null;
+        }
+        return new ServiceUnitStateData(state, broker);
+    }
+
+    private ServiceUnitStateData testValue0(String broker) {
+        ServiceUnitState to = nextValidState(testState0);
+        testState0 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue1(String broker) {
+        ServiceUnitState to = nextValidState(testState1);
+        testState1 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue2(String broker) {
+        ServiceUnitState to = nextValidState(testState2);
+        testState2 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue3(String broker) {
+        ServiceUnitState to = nextValidState(testState3);
+        testState3 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue4(String broker) {
+        ServiceUnitState to = nextValidState(testState4);
+        testState4 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitState nextValidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> to != Free && to != Splitting && isValidTransition(from, to))
+                .collect(Collectors.toList());
+        var state=  candidates.get(RANDOM.nextInt(candidates.size()));
+        System.out.println("test state:" + state);
+        return state;
+    }
+
+    private ServiceUnitState nextInvalidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> !isValidTransition(from, to))
+                .collect(Collectors.toList());
+        if (candidates.size() == 0) {
+            return null;
+        }
+        return candidates.get(RANDOM.nextInt(candidates.size()));
+    }
+
+    private List<ServiceUnitState> nextStatesToNull(ServiceUnitState from) {
+        if (from == null) {
+            return List.of();
+        }
+        return switch (from) {
+            case Assigned -> List.of(Owned);
+            case Owned -> List.of();
+            case Splitting -> List.of();
+            default -> List.of();
+        };
+    }
+
+    @BeforeMethod
+    @Override
+    public void setup() throws Exception {
+        super.internalSetup();
+
+        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+        admin.tenants().createTenant("my-property",
+                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use")));
+        admin.namespaces().createNamespace("my-property/use/my-ns");
+
+        compactionScheduler = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());
+        bk = pulsar.getBookKeeperClientFactory().create(this.conf, null, null, Optional.empty(), null);
+        schema = JSONSchema.of(ServiceUnitStateData.class);
+        strategy = new ServiceUnitStateCompactionStrategy();
+        strategy.checkBrokers(false);
+
+    }
+
+
+    @AfterMethod(alwaysRun = true)
+    @Override
+    public void cleanup() throws Exception {
+        super.internalCleanup();
+
+        if (compactionScheduler != null) {
+            compactionScheduler.shutdownNow();
+        }
+    }
+
+
+    public record TestData(
+            String topic,
+            Map<String, ServiceUnitStateData> expected,
+            List<Pair<String, ServiceUnitStateData>> all) {
+
+    }
+    TestData generateTestData() throws PulsarAdminException, PulsarClientException {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+        final int numMessages = 20;
+        final int maxKeys = 5;
+
+        // Configure retention to ensue data is retained for reader
+        admin.namespaces().setRetention("my-property/use/my-ns", new RetentionPolicies(-1, -1));
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+
+        Map<String, ServiceUnitStateData> expected = new HashMap<>();
+        List<Pair<String, ServiceUnitStateData>> all = new ArrayList<>();
+        Random r = new Random(0);
+
+        pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub1")
+                .readCompacted(true)
+                .subscribe().close();
+
+        for (int j = 0; j < numMessages; j++) {
+            int keyIndex = r.nextInt(maxKeys);
+            String key = "key" + keyIndex;
+            ServiceUnitStateData expectedData = expected.get(key);
+            ServiceUnitState state = expectedData == null ? nextValidState(Assigned) :
+                    r.nextInt(2) == 0 ? nextInvalidState(expectedData.state()) :
+                            nextValidState(expectedData.state());
+            ServiceUnitStateData value = new ServiceUnitStateData(state, key + ":" + j);
+            producer.newMessage().key(key).value(value).send();
+            ServiceUnitStateData prev = expected.get(key);
+            if (!strategy.shouldKeepLeft(prev, value)) {
+                expected.put(key, value);
+            }
+            all.add(Pair.of(key, value));
+        }
+        return new TestData(topic, expected, all);
+    }
+
+    @Test
+    public void testCompaction() throws Exception {
+        TestData testData = generateTestData();
+        var topic = testData.topic;
+        var expected = testData.expected;
+        var all = testData.all;
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic, false);
+        // Compacted topic ledger should have same number of entry equals to number of unique key.
+        //Assert.assertEquals(internalStats.compactedLedger.entries, expected.size());
+        Assert.assertTrue(internalStats.compactedLedger.ledgerId > -1);
+        Assert.assertFalse(internalStats.compactedLedger.offloaded);
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+                Assert.assertEquals(expected.remove(m.getKey()), m.getValue());
+                if (expected.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(expected.isEmpty());
+        }
+
+        // can get full backlog if read compacted disabled
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(false).subscribe()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+                Pair<String, ServiceUnitStateData> expectedMessage = all.remove(0);
+                Assert.assertEquals(expectedMessage.getLeft(), m.getKey());
+                Assert.assertEquals(expectedMessage.getRight(), m.getValue());
+                if (all.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(all.isEmpty());
+        }
+    }
+
+    @Test
+    public void testCompactionWithReader() throws Exception {
+        TestData testData = generateTestData();
+        var topic = testData.topic;
+        var expected = testData.expected;
+        var all = testData.all;
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Reader<ServiceUnitStateData> reader = pulsarClient.newReader(schema).topic(topic).readCompacted(true)
+                .startMessageId(MessageId.earliest).create()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = reader.readNext(2, TimeUnit.SECONDS);
+                Assert.assertEquals(expected.remove(m.getKey()), m.getValue());
+                if (expected.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(expected.isEmpty());
+        }
+
+        // can get full backlog if read compacted disabled
+        try (Reader<ServiceUnitStateData> reader = pulsarClient.newReader(schema).topic(topic).readCompacted(false)
+                .startMessageId(MessageId.earliest).create()) {
+            while (true) {
+                Message<ServiceUnitStateData> m = reader.readNext(2, TimeUnit.SECONDS);
+                Pair<String, ServiceUnitStateData> expectedMessage = all.remove(0);
+                Assert.assertEquals(expectedMessage.getLeft(), m.getKey());
+                Assert.assertEquals(expectedMessage.getRight(), m.getValue());
+                if (all.isEmpty()) {
+                    break;
+                }
+            }
+            Assert.assertTrue(all.isEmpty());
+        }
+    }
+
+
+    @Test
+    public void testCompactionWithTableview() throws Exception {
+        var tv = pulsar.getClient().newTableViewBuilder(schema)
+                .topic("persistent://my-property/use/my-ns/my-topic1")
+                .loadConf(Map.of(
+                        "topicCompactionStrategyClassName",
+                        ServiceUnitStateCompactionStrategy.class.getName()))
+                .create();
+
+        ((ServiceUnitStateCompactionStrategy)
+                FieldUtils.readDeclaredField(tv, "compactionStrategy", true))
+                .checkBrokers(false);
+        TestData testData = generateTestData();
+        var topic = testData.topic;
+        var expected = testData.expected;
+        var expectedCopy = new HashMap<>(expected);
+
+        Awaitility.await()
+                .pollInterval(200, TimeUnit.MILLISECONDS)
+                .atMost(10, TimeUnit.SECONDS)
+                .untilAsserted(() -> assertEquals(expectedCopy.size(), tv.size()));
+
+        for(var etr : tv.entrySet()){
+            Assert.assertEquals(expectedCopy.remove(etr.getKey()), etr.getValue());
+            if (expectedCopy.isEmpty()) {
+                break;
+            }
+        }
+
+        Assert.assertTrue(expectedCopy.isEmpty());
+        tv.close();;
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        var tableview = pulsar.getClient().newTableViewBuilder(schema)
+                .topic(topic)
+                .loadConf(Map.of(
+                        "topicCompactionStrategyClassName",
+                        ServiceUnitStateCompactionStrategy.class.getName()))
+                .create();
+        for(var etr : tableview.entrySet()){
+            Assert.assertEquals(expected.remove(etr.getKey()), etr.getValue());
+            if (expected.isEmpty()) {
+                break;
+            }
+        }
+        Assert.assertTrue(expected.isEmpty());
+        tableview.close();
+
+    }
+
+
+    @Test
+    public void testReadCompactedBeforeCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content0");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content1");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+    }
+
+    @Test
+    public void testReadEntriesAfterCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        producer.newMessage().key("key0").value(testValue0("content3")).send();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content3");
+        }
+    }
+
+    @Test
+    public void testSeekEarliestAfterCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            consumer.seek(MessageId.earliest);
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(false).subscribe()) {
+            consumer.seek(MessageId.earliest);
+
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content0");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content1");
+
+            m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+    }
+
+    @Test
+    public void testBrokerRestartAfterCompaction() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+        producer.newMessage().key("key0").value(testValue0("content1")).send();
+        producer.newMessage().key("key0").value(testValue0( "content2")).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+
+        stopBroker();
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            consumer.receive();
+            Assert.fail("Shouldn't have been able to receive anything");
+        } catch (PulsarClientException e) {
+            // correct behaviour
+        }
+        startBroker();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content2");
+        }
+    }
+
+    @Test
+    public void testCompactEmptyTopic() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema)
+                .topic(topic)
+                .enableBatching(true)
+                .create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        producer.newMessage().key("key0").value(testValue0( "content0")).send();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive();
+            Assert.assertEquals(m.getKey(), "key0");
+            Assert.assertEquals(m.getValue().broker(), "content0");
+        }
+    }
+
+    @Test
+    public void testWholeBatchCompactedOut() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        // subscribe before sending anything, so that we get all messages
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe().close();
+
+        try (Producer<ServiceUnitStateData> producerNormal = pulsarClient.newProducer(schema).topic(topic)
+                .enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+             Producer<ServiceUnitStateData> producerBatch = pulsarClient.newProducer(schema).topic(topic)
+                     .maxPendingMessages(3)
+                     .enableBatching(true)
+                     .batchingMaxMessages(3)
+                     .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+                     .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                     .create()) {
+            producerBatch.newMessage().key("key1").value(testValue1("my-message-1")).sendAsync();
+            producerBatch.newMessage().key("key1").value(testValue1( "my-message-2")).sendAsync();
+            producerBatch.newMessage().key("key1").value(testValue1("my-message-3")).sendAsync();
+            producerNormal.newMessage().key("key1").value(testValue1( "my-message-4")).send();
+        }
+
+        // compact the topic
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic)
+                .subscriptionName("sub1").readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> message = consumer.receive();
+            Assert.assertEquals(message.getKey(), "key1");
+            Assert.assertEquals(new String(message.getValue().broker()), "my-message-4");
+        }
+    }
+
+    public void testCompactionWithLastDeletedKey() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema).topic(topic).enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition).create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("1").value(testValue(Owned, "1")).send();
+        producer.newMessage().key("2").value(testValue(Owned, "3")).send();
+        producer.newMessage().key("3").value(testValue(Owned, "5")).send();
+        producer.newMessage().key("1").value(null).send();
+        producer.newMessage().key("2").value(null).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        Set<String> expected = Sets.newHashSet("3");
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+            assertTrue(expected.remove(m.getKey()));
+        }
+    }
+
+    @Test(timeOut = 20000)
+    public void testEmptyCompactionLedger() throws Exception {
+        String topic = "persistent://my-property/use/my-ns/my-topic1";
+
+        Producer<ServiceUnitStateData> producer = pulsarClient.newProducer(schema).topic(topic).enableBatching(true)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition).create();
+
+        pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1").readCompacted(true).subscribe().close();
+
+        producer.newMessage().key("1").value(testValue(Owned, "1")).send();
+        producer.newMessage().key("2").value(testValue(Owned, "3")).send();
+        producer.newMessage().key("1").value(null).send();
+        producer.newMessage().key("2").value(null).send();
+
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(m);
+        }
+    }
+
+    @Test(timeOut = 20000)
+    public void testAllEmptyCompactionLedger() throws Exception {
+        final String topic =
+                "persistent://my-property/use/my-ns/testAllEmptyCompactionLedger" + UUID.randomUUID().toString();
+
+        final int messages = 10;
+
+        // 1.create producer and publish message to the topic.
+        ProducerBuilder<ServiceUnitStateData> builder = pulsarClient.newProducer(schema).topic(topic);
+        builder.batchingMaxMessages(messages / 5);
+
+        Producer<ServiceUnitStateData> producer = builder.create();
+
+        List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key("1").value(null).sendAsync());
+        }
+
+        FutureUtil.waitForAll(futures).get();
+
+        // 2.compact the topic.
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // consumer with readCompacted enabled only get compacted entries
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+            Message<ServiceUnitStateData> m = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(m);
+        }
+    }
+
+    @Test(timeOut = 20000)
+    public void testCompactMultipleTimesWithoutEmptyMessage()
+            throws PulsarClientException, ExecutionException, InterruptedException {
+        final String topic =
+                "persistent://my-property/use/my-ns/testCompactMultipleTimesWithoutEmptyMessage" + UUID.randomUUID()
+                        .toString();
+
+        final int messages = 10;
+        final String key = "1";
+
+        // 1.create producer and publish message to the topic.
+        ProducerBuilder<ServiceUnitStateData> builder = pulsarClient.newProducer(schema).topic(topic);
+        builder.enableBatching(true);
+
+
+        Producer<ServiceUnitStateData> producer = builder.create();
+
+        List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + ""))).sendAsync());
+        }
+
+        FutureUtil.waitForAll(futures).get();
+
+        // 2.compact the topic.
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // 3. Send more ten messages
+        futures.clear();
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + 10 + ""))).sendAsync());
+        }
+        FutureUtil.waitForAll(futures).get();
+
+        // 4.compact again.
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema).topic(topic).subscriptionName("sub1")
+                .readCompacted(true).subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe()) {
+            Message<ServiceUnitStateData> m1 = consumer.receive();
+            assertNotNull(m1);
+            assertEquals(m1.getKey(), key);
+            assertEquals(m1.getValue().broker(), "19");
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+    }
+
+    @Test(timeOut = 200000)
+    public void testReadUnCompacted()
+            throws PulsarClientException, ExecutionException, InterruptedException {
+        final String topic = "persistent://my-property/use/my-ns/testReadUnCompacted" + UUID.randomUUID().toString();
+
+        final int messages = 10;
+        final String key = "1";
+
+        // 1.create producer and publish message to the topic.
+        ProducerBuilder<ServiceUnitStateData> builder = pulsarClient.newProducer(schema).topic(topic);
+        builder.batchingMaxMessages(messages / 5);
+
+        Producer<ServiceUnitStateData> producer = builder.create();
+
+        List<CompletableFuture<MessageId>> futures = new ArrayList<>(messages);
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + ""))).sendAsync());
+        }
+
+        FutureUtil.waitForAll(futures).get();
+
+        // 2.compact the topic.
+        StrategicTwoPhaseCompactor compactor
+                = new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
+        compactor.compact(topic, strategy).get();
+
+        // 3. Send more ten messages
+        futures.clear();
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + 10 + ""))).sendAsync());
+        }
+        FutureUtil.waitForAll(futures).get();
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub1")
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe()) {
+            for (int i = 0; i < 11; i++) {
+                Message<ServiceUnitStateData> received = consumer.receive();
+                assertNotNull(received);
+                assertEquals(received.getKey(), key);
+                assertEquals(received.getValue().broker(), i + 9 + "");
+                consumer.acknowledge(received);
+            }
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+
+        // 4.Send empty message to delete the key-value in the compacted topic.
+        for (ServiceUnitState state : nextStatesToNull(testState0)) {
+            producer.newMessage().key(key).value(new ServiceUnitStateData(state, "xx")).send();
+        }
+        producer.newMessage().key(key).value(null).send();
+
+        // 5.compact the topic.
+        compactor.compact(topic, strategy).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub2")
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe()) {
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+
+        for (int i = 0; i < messages; i++) {
+            futures.add(producer.newMessage().key(key).value(testValue0((i + 20 + ""))).sendAsync());
+        }
+        FutureUtil.waitForAll(futures).get();
+
+        try (Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(topic)
+                .subscriptionName("sub3")
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe()) {
+            for (int i = 0; i < 10; i++) {
+                Message<ServiceUnitStateData> received = consumer.receive();
+                assertNotNull(received);
+                assertEquals(received.getKey(), key);
+                assertEquals(received.getValue().broker(), i + 20 + "");
+                consumer.acknowledge(received);
+            }
+            Message<ServiceUnitStateData> none = consumer.receive(2, TimeUnit.SECONDS);
+            assertNull(none);
+        }
+    }
+
+    @SneakyThrows
+    @Test
+    public void testHealthCheckTopicNotCompacted() {
+        NamespaceName heartbeatNamespaceV1 =
+                NamespaceService.getHeartbeatNamespace(pulsar.getAdvertisedAddress(), pulsar.getConfiguration());
+        String topicV1 = "persistent://" + heartbeatNamespaceV1.toString() + "/healthcheck";
+        NamespaceName heartbeatNamespaceV2 =
+                NamespaceService.getHeartbeatNamespaceV2(pulsar.getAdvertisedAddress(), pulsar.getConfiguration());
+        String topicV2 = heartbeatNamespaceV2.toString() + "/healthcheck";
+        Producer<ServiceUnitStateData> producer1 = pulsarClient.newProducer(schema).topic(topicV1).create();
+        Producer<ServiceUnitStateData> producer2 = pulsarClient.newProducer(schema).topic(topicV2).create();
+        Optional<Topic> topicReferenceV1 = pulsar.getBrokerService().getTopic(topicV1, false).join();
+        Optional<Topic> topicReferenceV2 = pulsar.getBrokerService().getTopic(topicV2, false).join();
+        assertFalse(((SystemTopic) topicReferenceV1.get()).isCompactionEnabled());
+        assertFalse(((SystemTopic) topicReferenceV2.get()).isCompactionEnabled());
+        producer1.close();
+        producer2.close();
+    }
+
+    @Test(timeOut = 60000)
+    public void testCompactionWithMarker() throws Exception {
+        String namespace = "my-property/use/my-ns";
+        final TopicName dest = TopicName.get(
+                BrokerTestUtil.newUniqueName("persistent://" + namespace + "/testWriteMarker"));
+        admin.topics().createNonPartitionedTopic(dest.toString());
+        @Cleanup
+        Consumer<ServiceUnitStateData> consumer = pulsarClient.newConsumer(schema)
+                .topic(dest.toString())
+                .subscriptionName("test-compaction-sub")
+                .subscriptionType(SubscriptionType.Exclusive)
+                .readCompacted(true)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
+                .subscribe();

Review Comment:
   The consumer is useless.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1091442850


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -202,6 +205,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(COMPACTION_NEVER_RUN);
     private final CompactedTopic compactedTopic;
 
+    // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users.
+    private static Map<String, TopicCompactionStrategy> strategicCompactionMap = Map.of(
+            ServiceUnitStateChannelImpl.TOPIC,
+            new ServiceUnitStateCompactionStrategy());

Review Comment:
   Regarding adding strategy in topic policy, I'm wondering that’s over-engineering at this point since we haven’t agreed to expose this to users. 
   
   Also, another property in topic policy would multiply as many as topics.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] codelipenghui commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.

codelipenghui commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1090373736


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1563,6 +1571,8 @@ public void checkCompaction() {
                 }
 
                 if (backlogEstimate > compactionThreshold) {
+                    log.info("topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering compaction",

Review Comment:
   It should be a debug-level log. 



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -202,6 +205,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(COMPACTION_NEVER_RUN);
     private final CompactedTopic compactedTopic;
 
+    // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users.
+    private static Map<String, TopicCompactionStrategy> strategicCompactionMap = Map.of(
+            ServiceUnitStateChannelImpl.TOPIC,
+            new ServiceUnitStateCompactionStrategy());

Review Comment:
   Is it better to use `Optional<TopicCompactionStrategy> compactionStrategy` and set the value to `ServiceUnitStateCompactionStrategy` if the topic is `ServiceUnitStateChannelImpl.TOPIC`. Maintaining a map with the topic name as the key in a `PersistentTopic` instance is a little confusing.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] eolivelli commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "eolivelli (via GitHub)" <gi...@apache.org>.

eolivelli commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1083715660


##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -385,6 +385,7 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
                             promise.completeExceptionally(e);
                             return;
                         }
+                        outstanding.release(MAX_OUTSTANDING);

Review Comment:
   should we move this after `promise.complete(null);` ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1090442037


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1563,6 +1571,8 @@ public void checkCompaction() {
                 }
 
                 if (backlogEstimate > compactionThreshold) {
+                    log.info("topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering compaction",

Review Comment:
   I thought this topic compaction log is useful. Could you share your concerns about this?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1091501985


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1563,6 +1571,8 @@ public void checkCompaction() {
                 }
 
                 if (backlogEstimate > compactionThreshold) {
+                    log.info("topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering compaction",

Review Comment:
   Updated.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] Technoboy- commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.

Technoboy- commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1089589647


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -619,14 +618,13 @@ private void doCleanup(String broker) {
                     || StringUtils.equals(broker, stateData.sourceBroker())) {
                 log.info("Cleaning ownership serviceUnit:{}, stateData:{}.", serviceUnit, stateData);
                 tombstoneAsync(serviceUnit).whenComplete((__, e) -> {
-                    if (e == null) {
-                        serviceUnitTombstoneCnt.incrementAndGet();
-                    } else {
+                    if (e != null) {
                         log.error("Failed cleaning the ownership serviceUnit:{}, stateData:{}.",
                                 serviceUnit, stateData);
                         serviceUnitTombstoneErrorCnt.incrementAndGet();
                     }
                 });
+                serviceUnitTombstoneCnt.incrementAndGet();

Review Comment:
   Could you help explain why do we need modify here ?
   we update `serviceUnitTombstoneErrorCnt` and `serviceUnitTombstoneCnt` in the async method. these values could both be 0. and let line 643 pass.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] codecov-commenter commented on pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by GitBox <gi...@apache.org>.

codecov-commenter commented on PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#issuecomment-1397872279

   # [Codecov](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
   > Merging [#19045](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (e348be3) into [master](https://codecov.io/gh/apache/pulsar/commit/d8569cd4ec6da14f8b2b9338db1ed2f6a3eacf0a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (d8569cd) will **decrease** coverage by `23.18%`.
   > The diff coverage is `12.50%`.
   
   [![Impacted file tree graph](https://codecov.io/gh/apache/pulsar/pull/19045/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
   
   ```diff
   @@              Coverage Diff              @@
   ##             master   #19045       +/-   ##
   =============================================
   - Coverage     47.20%   24.02%   -23.18%     
   + Complexity    10645      295    -10350     
   =============================================
     Files           709     1642      +933     
     Lines         69421   125195    +55774     
     Branches       7449    13663     +6214     
   =============================================
   - Hits          32769    30082     -2687     
   - Misses        32984    90683    +57699     
   - Partials       3668     4430      +762     
   ```
   
   | Flag | Coverage Δ | |
   |---|---|---|
   | inttests | `24.02% <12.50%> (?)` | |
   | unittests | `?` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Impacted Files](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
   |---|---|---|
   | [...xtensions/channel/ServiceUnitStateChannelImpl.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9sb2FkYmFsYW5jZS9leHRlbnNpb25zL2NoYW5uZWwvU2VydmljZVVuaXRTdGF0ZUNoYW5uZWxJbXBsLmphdmE=) | `0.99% <0.00%> (+0.99%)` | :arrow_up: |
   | [.../pulsar/compaction/StrategicTwoPhaseCompactor.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbXBhY3Rpb24vU3RyYXRlZ2ljVHdvUGhhc2VDb21wYWN0b3IuamF2YQ==) | `0.00% <0.00%> (-76.20%)` | :arrow_down: |
   | [...ns/channel/ServiceUnitStateCompactionStrategy.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9sb2FkYmFsYW5jZS9leHRlbnNpb25zL2NoYW5uZWwvU2VydmljZVVuaXRTdGF0ZUNvbXBhY3Rpb25TdHJhdGVneS5qYXZh) | `13.33% <13.33%> (ø)` | |
   | [...sar/broker/service/persistent/PersistentTopic.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvUGVyc2lzdGVudFRvcGljLmphdmE=) | `46.93% <16.66%> (-14.63%)` | :arrow_down: |
   | [...n/java/org/apache/pulsar/broker/PulsarService.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9QdWxzYXJTZXJ2aWNlLmphdmE=) | `53.80% <25.00%> (-4.25%)` | :arrow_down: |
   | [...java/org/apache/pulsar/proxy/stats/TopicStats.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLXByb3h5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvcHJveHkvc3RhdHMvVG9waWNTdGF0cy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
   | [...ava/org/apache/pulsar/broker/admin/v1/Brokers.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92MS9Ccm9rZXJzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
   | [...va/org/apache/pulsar/broker/admin/v1/Clusters.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92MS9DbHVzdGVycy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
   | [.../org/apache/pulsar/broker/admin/v1/Properties.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92MS9Qcm9wZXJ0aWVzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
   | [.../apache/pulsar/broker/admin/v2/ResourceGroups.java](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9SZXNvdXJjZUdyb3Vwcy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
   | ... and [1406 more](https://codecov.io/gh/apache/pulsar/pull/19045?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] eolivelli commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "eolivelli (via GitHub)" <gi...@apache.org>.

eolivelli commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1083715886


##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java:
##########
@@ -0,0 +1,833 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.isValidTransition;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateCompactionStrategy;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+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.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-compaction")
+public class ServiceUnitStateCompactionTest extends MockedPulsarServiceBaseTest {
+    private ScheduledExecutorService compactionScheduler;
+    private BookKeeper bk;
+    private JSONSchema<ServiceUnitStateData> schema;
+    private ServiceUnitStateCompactionStrategy strategy;
+
+    private ServiceUnitState testState0 = Free;
+    private ServiceUnitState testState1 = Free;
+    private ServiceUnitState testState2 = Free;
+    private ServiceUnitState testState3 = Free;
+    private ServiceUnitState testState4 = Free;
+
+    private static Random RANDOM = new Random();
+
+
+    private ServiceUnitStateData testValue(ServiceUnitState state, String broker) {
+        if (state == Free) {
+            return null;
+        }
+        return new ServiceUnitStateData(state, broker);
+    }
+
+    private ServiceUnitStateData testValue0(String broker) {
+        ServiceUnitState to = nextValidState(testState0);
+        testState0 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue1(String broker) {
+        ServiceUnitState to = nextValidState(testState1);
+        testState1 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue2(String broker) {
+        ServiceUnitState to = nextValidState(testState2);
+        testState2 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue3(String broker) {
+        ServiceUnitState to = nextValidState(testState3);
+        testState3 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue4(String broker) {
+        ServiceUnitState to = nextValidState(testState4);
+        testState4 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitState nextValidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> to != Free && to != Splitting && isValidTransition(from, to))
+                .collect(Collectors.toList());
+        var state=  candidates.get(RANDOM.nextInt(candidates.size()));
+        System.out.println("test state:" + state);
+        return state;
+    }
+
+    private ServiceUnitState nextInvalidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> !isValidTransition(from, to))
+                .collect(Collectors.toList());
+        if (candidates.size() == 0) {
+            return null;
+        }
+        return candidates.get(RANDOM.nextInt(candidates.size()));
+    }
+
+    private List<ServiceUnitState> nextStatesToNull(ServiceUnitState from) {
+        if (from == null) {
+            return List.of();
+        }
+        return switch (from) {
+            case Assigned -> List.of(Owned);
+            case Owned -> List.of();
+            case Splitting -> List.of();
+            default -> List.of();
+        };
+    }
+
+    @BeforeMethod
+    @Override
+    public void setup() throws Exception {
+        super.internalSetup();
+
+        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+        admin.tenants().createTenant("my-property",
+                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use")));
+        admin.namespaces().createNamespace("my-property/use/my-ns");
+
+        compactionScheduler = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());
+        bk = pulsar.getBookKeeperClientFactory().create(this.conf, null, null, Optional.empty(), null);
+        schema = JSONSchema.of(ServiceUnitStateData.class);

Review Comment:
   Schema.JSON



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1084364025


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategy.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.extensions.channel;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Released;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.topics.TopicCompactionStrategy;
+
+public class ServiceUnitStateCompactionStrategy implements TopicCompactionStrategy<ServiceUnitStateData> {
+
+    private final Schema<ServiceUnitStateData> schema;
+
+    private boolean checkBrokers = true;
+
+    public ServiceUnitStateCompactionStrategy() {
+        schema = JSONSchema.of(ServiceUnitStateData.class);

Review Comment:
   Updated



##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java:
##########
@@ -0,0 +1,833 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigned;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.isValidTransition;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateCompactionStrategy;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+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.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-compaction")
+public class ServiceUnitStateCompactionTest extends MockedPulsarServiceBaseTest {
+    private ScheduledExecutorService compactionScheduler;
+    private BookKeeper bk;
+    private JSONSchema<ServiceUnitStateData> schema;
+    private ServiceUnitStateCompactionStrategy strategy;
+
+    private ServiceUnitState testState0 = Free;
+    private ServiceUnitState testState1 = Free;
+    private ServiceUnitState testState2 = Free;
+    private ServiceUnitState testState3 = Free;
+    private ServiceUnitState testState4 = Free;
+
+    private static Random RANDOM = new Random();
+
+
+    private ServiceUnitStateData testValue(ServiceUnitState state, String broker) {
+        if (state == Free) {
+            return null;
+        }
+        return new ServiceUnitStateData(state, broker);
+    }
+
+    private ServiceUnitStateData testValue0(String broker) {
+        ServiceUnitState to = nextValidState(testState0);
+        testState0 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue1(String broker) {
+        ServiceUnitState to = nextValidState(testState1);
+        testState1 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue2(String broker) {
+        ServiceUnitState to = nextValidState(testState2);
+        testState2 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue3(String broker) {
+        ServiceUnitState to = nextValidState(testState3);
+        testState3 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitStateData testValue4(String broker) {
+        ServiceUnitState to = nextValidState(testState4);
+        testState4 = to;
+        return testValue(to, broker);
+    }
+
+    private ServiceUnitState nextValidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> to != Free && to != Splitting && isValidTransition(from, to))
+                .collect(Collectors.toList());
+        var state=  candidates.get(RANDOM.nextInt(candidates.size()));
+        System.out.println("test state:" + state);
+        return state;
+    }
+
+    private ServiceUnitState nextInvalidState(ServiceUnitState from) {
+        List<ServiceUnitState> candidates = Arrays.stream(ServiceUnitState.values())
+                .filter(to -> !isValidTransition(from, to))
+                .collect(Collectors.toList());
+        if (candidates.size() == 0) {
+            return null;
+        }
+        return candidates.get(RANDOM.nextInt(candidates.size()));
+    }
+
+    private List<ServiceUnitState> nextStatesToNull(ServiceUnitState from) {
+        if (from == null) {
+            return List.of();
+        }
+        return switch (from) {
+            case Assigned -> List.of(Owned);
+            case Owned -> List.of();
+            case Splitting -> List.of();
+            default -> List.of();
+        };
+    }
+
+    @BeforeMethod
+    @Override
+    public void setup() throws Exception {
+        super.internalSetup();
+
+        admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+        admin.tenants().createTenant("my-property",
+                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use")));
+        admin.namespaces().createNamespace("my-property/use/my-ns");
+
+        compactionScheduler = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());
+        bk = pulsar.getBookKeeperClientFactory().create(this.conf, null, null, Optional.empty(), null);
+        schema = JSONSchema.of(ServiceUnitStateData.class);

Review Comment:
   updated.



##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java:
##########
@@ -385,6 +385,7 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader,
                             promise.completeExceptionally(e);
                             return;
                         }
+                        outstanding.release(MAX_OUTSTANDING);

Review Comment:
   No, I found a deadlock if we don't release this semaphore before completing the future.
   
   deadLock example
   
   ```
   import java.util.concurrent.CompletableFuture;
   import java.util.concurrent.ExecutorService;
   import java.util.concurrent.Executors;
   import java.util.concurrent.Semaphore;
   import org.junit.Test;
   
   public class SemaphoreTest {
       
       @Test
       public void deadlockTest()  {
           Semaphore sm = new Semaphore(2);
   
           ExecutorService executor = Executors.newSingleThreadExecutor();
           CompletableFuture<String> future = new CompletableFuture();
           CompletableFuture.supplyAsync(() -> {
               System.out.println("starting");
               try {
                   sm.acquire(2);
                   System.out.println("acquired 2");
               } catch (InterruptedException e) {
               }
               try {
                   Thread.sleep(1000*5);
               } catch (InterruptedException e) {
               }
               System.out.println("return!");
               future.complete(""); // hangs here.
               System.out.println("proceeding");
               sm.release(2);
               System.out.println("released 2");
               return "";
           }, executor);
           future.thenCompose(x -> {
               try {
                   System.out.println("acquiring one!");
                   sm.acquire(); // hangs here.
                   System.out.println("acquired 1");
                   sm.release();
               } catch (InterruptedException e) {
               }
               return CompletableFuture.completedFuture("");
           }).join();
       }
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1089662048


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -619,14 +618,13 @@ private void doCleanup(String broker) {
                     || StringUtils.equals(broker, stateData.sourceBroker())) {
                 log.info("Cleaning ownership serviceUnit:{}, stateData:{}.", serviceUnit, stateData);
                 tombstoneAsync(serviceUnit).whenComplete((__, e) -> {
-                    if (e == null) {
-                        serviceUnitTombstoneCnt.incrementAndGet();
-                    } else {
+                    if (e != null) {
                         log.error("Failed cleaning the ownership serviceUnit:{}, stateData:{}.",
                                 serviceUnit, stateData);
                         serviceUnitTombstoneErrorCnt.incrementAndGet();
                     }
                 });
+                serviceUnitTombstoneCnt.incrementAndGet();

Review Comment:
   It relies on `producer.flush()` to persist all outstanding messages before returning this call. 
   
   Yes, because these values could be 0, I updated this code path here. Also, I further cleaned the metrics computation code to be cleaner.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] codelipenghui commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.

codelipenghui commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1091451187


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -202,6 +205,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(COMPACTION_NEVER_RUN);
     private final CompactedTopic compactedTopic;
 
+    // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users.
+    private static Map<String, TopicCompactionStrategy> strategicCompactionMap = Map.of(
+            ServiceUnitStateChannelImpl.TOPIC,
+            new ServiceUnitStateCompactionStrategy());

Review Comment:
   Ah ok, that makes sense to me.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] heesung-sn commented on a diff in pull request #19045: [improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy

Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.

heesung-sn commented on code in PR #19045:
URL: https://github.com/apache/pulsar/pull/19045#discussion_r1091441218


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1563,6 +1571,8 @@ public void checkCompaction() {
                 }
 
                 if (backlogEstimate > compactionThreshold) {
+                    log.info("topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering compaction",

Review Comment:
   Ok. I will make this debug lvl.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org