resolved comments

Author: heesung-sn

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

@@ -39,7 +39,6 @@
 import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.MetadataState.Stable;
 import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.MetadataState.Unstable;
 import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData.state;
-import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData.versionId;
 import static org.apache.pulsar.metadata.api.extended.SessionEvent.SessionLost;
 import static org.apache.pulsar.metadata.api.extended.SessionEvent.SessionReestablished;
 import com.google.common.annotations.VisibleForTesting;
@@ -104,7 +103,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
             NamespaceName.SYSTEM_NAMESPACE,
             "loadbalancer-service-unit-state").toString();
     private static final long MAX_IN_FLIGHT_STATE_WAITING_TIME_IN_MILLIS = 30 * 1000; // 30sec
-    private static final long VERSION_ID_INIT = 1; // initial versionId
+    public static final long VERSION_ID_INIT = 1; // initial versionId
     private static final long OWNERSHIP_MONITOR_DELAY_TIME_IN_SECS = 60;
     public static final long MAX_CLEAN_UP_DELAY_TIME_IN_SECS = 3 * 60; // 3 mins
     private static final long MIN_CLEAN_UP_DELAY_TIME_IN_SECS = 0; // 0 secs to clean immediately
@@ -454,11 +453,11 @@ public CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit) {
 
     private long getNextVersionId(String serviceUnit) {
         var data = tableview.get(serviceUnit);
-        return versionId(data) + 1;
+        return getNextVersionId(data);
     }
 
     private long getNextVersionId(ServiceUnitStateData data) {
-        return versionId(data) + 1;
+        return data == null ? VERSION_ID_INIT : data.versionId() + 1;
     }
 
     public CompletableFuture<String> publishAssignEventAsync(String serviceUnit, String broker) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateData.java

@@ -51,8 +51,4 @@ public ServiceUnitStateData(ServiceUnitState state, String broker, boolean force
     public static ServiceUnitState state(ServiceUnitStateData data) {
         return data == null ? ServiceUnitState.Init : data.state();
     }
-
-    public static long versionId(ServiceUnitStateData data) {
-        return data == null ? 0 : data.versionId();
-    }
 }

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

@@ -26,7 +26,6 @@
 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.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData.state;
-import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData.versionId;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
@@ -55,6 +54,7 @@
 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.ServiceUnitStateChannelImpl;
 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;
@@ -953,4 +953,8 @@ public void testReadUnCompacted()
             assertNull(none);
         }
     }
+
+    public static long versionId(ServiceUnitStateData data) {
+        return data == null ? ServiceUnitStateChannelImpl.VERSION_ID_INIT - 1 : data.versionId();
+    }
 }