New snapshot

pom.xml

@@ -2,8 +2,8 @@
 <project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xmlns="http://maven.apache.org/POM/4.0.0"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
     <modelVersion>4.0.0</modelVersion>
-
     <groupId>com.alipay.sofa</groupId>
     <artifactId>registry-parent</artifactId>
     <version>5.4.0-SNAPSHOT</version>
@@ -75,7 +75,7 @@
         <lookout.version>1.5.2</lookout.version>
         <mockito.version>1.10.19</mockito.version>
         <powermock.version>1.6.6</powermock.version>
-        <jraft.version>1.2.5</jraft.version>
+        <jraft.version>1.2.7.beta1</jraft.version>
         <metrics.version>4.0.2</metrics.version>
         <commons-io.version>2.4</commons-io.version>
         <jetty.version>[9.4.17.v20190418,9.4.19.v20190610]</jetty.version>
@@ -85,6 +85,11 @@
 
     <dependencyManagement>
         <dependencies>
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>jul-to-slf4j</artifactId>
+                <version>1.7.9</version>
+            </dependency>
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-starter-parent</artifactId>

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/dataserver/Datum.java

@@ -62,7 +62,7 @@ public Datum() {
      * @param dataCenter
      */
     public Datum(String dataInfoId, String dataCenter) {
-        this.dataInfoId = dataInfoId;
+        this.dataInfoId = WordCache.getInstance().getWordCache(dataInfoId);
         this.dataCenter = WordCache.getInstance().getWordCache(dataCenter);
         updateVersion();
     }
@@ -244,7 +244,7 @@ public void setContainsUnPub(boolean containsUnPub) {
         this.containsUnPub = containsUnPub;
     }
 
-    public static Datum processDatum(Datum datum) {
+    public static Datum internDatum(Datum datum) {
         datum.setDataCenter(datum.getDataCenter());
         datum.setDataInfoId(datum.getDataInfoId());
         datum.setDataId(datum.getDataId());
@@ -253,7 +253,13 @@ public static Datum processDatum(Datum datum) {
 
         Map<String, Publisher> pubMap = datum.getPubMap();
         if (pubMap != null && !pubMap.isEmpty()) {
-            pubMap.forEach((registerId, publisher) -> Publisher.processPublisher(publisher));
+            pubMap.forEach((registerId, publisher) -> {
+                // let registerId == pub.getRegisterId in every <registerId, pub>, for reducing old gen memory
+                // because this Datum is put into Memory directly, by DatumCache.coverDatum
+                publisher.setRegisterId(registerId);
+                // change publisher word cache
+                Publisher.internPublisher(publisher);
+            });
         }
 
         return datum;

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/BaseInfo.java

@@ -16,12 +16,12 @@
  */
 package com.alipay.sofa.registry.common.model.store;
 
-import com.fasterxml.jackson.annotation.JsonIgnore;
-
 import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
 /**
  *
  * @author shangyu.wh
@@ -136,7 +136,7 @@ public String getProcessId() {
      * @param processId  value to be assigned to property processId
      */
     public void setProcessId(String processId) {
-        this.processId = processId;
+        this.processId = WordCache.getInstance().getWordCache(processId);
     }
 
     /**
@@ -273,7 +273,7 @@ public String getClientId() {
      * @param clientId  value to be assigned to property clientId
      */
     public void setClientId(String clientId) {
-        this.clientId = clientId;
+        this.clientId = WordCache.getInstance().getWordCache(clientId);
     }
 
     /**

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/Publisher.java

@@ -16,12 +16,12 @@
  */
 package com.alipay.sofa.registry.common.model.store;
 
+import java.util.List;
+
 import com.alipay.sofa.registry.common.model.PublishType;
 import com.alipay.sofa.registry.common.model.ServerDataBox;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 
-import java.util.List;
-
 /**
  *
  * @author shangyu.wh
@@ -90,8 +90,8 @@ protected String getOtherInfo() {
      * @param publisher
      * @return
      */
-    public static Publisher processPublisher(Publisher publisher) {
-
+    public static Publisher internPublisher(Publisher publisher) {
+        publisher.setRegisterId(publisher.getRegisterId());
         publisher.setDataInfoId(publisher.getDataInfoId());
         publisher.setInstanceId(publisher.getInstanceId());
         publisher.setGroup(publisher.getGroup());
@@ -101,13 +101,6 @@ public static Publisher processPublisher(Publisher publisher) {
         publisher.setProcessId(publisher.getProcessId());
         publisher.setAppName(publisher.getAppName());
 
-        if (publisher.getSourceAddress() != null) {
-            publisher.setSourceAddress(new URL(publisher.getSourceAddress().getIpAddress(),
-                publisher.getSourceAddress().getPort()));
-        }
-
-        publisher.setAttributes(publisher.getAttributes());
-
         return publisher;
     }
 

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/Subscriber.java

@@ -16,13 +16,14 @@
  */
 package com.alipay.sofa.registry.common.model.store;
 
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 import com.alipay.sofa.registry.common.model.ElementType;
+import com.alipay.sofa.registry.common.model.constants.ValueConstants;
 import com.alipay.sofa.registry.core.model.ScopeEnum;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
 /**
  *
  * @author shangyu.wh
@@ -31,16 +32,16 @@
 public class Subscriber extends BaseInfo {
 
     /** UID */
-    private static final long                serialVersionUID = 98433360274932292L;
+    private static final long                       serialVersionUID = 98433360274932292L;
     /** */
-    private ScopeEnum                        scope;
+    private ScopeEnum                               scope;
     /** */
-    private ElementType                      elementType;
+    private ElementType                             elementType;
 
     /**
-     * all dataCenter push dataInfo version
+     * last push context
      */
-    private Map<String/*dataCenter*/, Long> lastPushVersions = new ConcurrentHashMap<>();
+    private Map<String/*dataCenter*/, PushContext> lastPushContexts = new ConcurrentHashMap<>();
 
     /**
      * Getter method for property <tt>scope</tt>.
@@ -71,7 +72,11 @@ public ElementType getElementType() {
      */
     public boolean checkVersion(String dataCenter, Long version) {
 
-        Long oldVersion = lastPushVersions.get(dataCenter);
+        PushContext lastPushContext = lastPushContexts.get(dataCenter);
+        if (lastPushContext == null) {
+            return version != null;
+        }
+        Long oldVersion = lastPushContext.pushVersion;
         if (oldVersion == null) {
             return version != null;
         } else {
@@ -88,15 +93,26 @@ public boolean checkVersion(String dataCenter, Long version) {
      * @return
      */
     public void checkAndUpdateVersion(String dataCenter, Long version) {
+        checkAndUpdateVersion(dataCenter, version, -1);
+    }
+
+    /**
+     * check version input greater or equal to current version
+     * @param version
+     * @return
+     */
+    public void checkAndUpdateVersion(String dataCenter, Long version, int pubCount) {
 
         while (true) {
-            Long oldVersion = lastPushVersions.putIfAbsent(dataCenter, version);
+            PushContext pushContext = new PushContext(version, pubCount);
+            PushContext oldPushContext = lastPushContexts.putIfAbsent(dataCenter, pushContext);
             // Add firstly
-            if (oldVersion == null) {
+            if (oldPushContext == null) {
                 break;
             } else {
-                if (version > oldVersion) {
-                    if (lastPushVersions.replace(dataCenter, oldVersion, version)) {
+                if (oldPushContext.pushVersion == null
+                    || (pushContext.pushVersion != null && pushContext.pushVersion > oldPushContext.pushVersion)) {
+                    if (lastPushContexts.replace(dataCenter, oldPushContext, pushContext)) {
                         break;
                     }
                 } else {
@@ -106,6 +122,23 @@ public void checkAndUpdateVersion(String dataCenter, Long version) {
         }
     }
 
+    /**
+     * If the pushed data is empty, check the last push, for avoid continuous empty datum push
+     */
+    public boolean allowPush(String dataCenter, int pubCount) {
+        boolean allowPush = true;
+        // condition of no push:
+        // 1. last push count is 0 and this time is also 0
+        // 2. last push is a valid push (version > 1)
+        if (pubCount == 0) {
+            PushContext pushContext = lastPushContexts.get(dataCenter);
+            allowPush = !(pushContext != null && pushContext.pushPubCount == 0
+            //last push is a valid push
+                          && pushContext.pushVersion != null && pushContext.pushVersion > ValueConstants.DEFAULT_NO_DATUM_VERSION);
+        }
+        return allowPush;
+    }
+
     /**
      * Setter method for property <tt>elementType</tt>.
      *
@@ -126,28 +159,10 @@ protected String getOtherInfo() {
         final StringBuilder sb = new StringBuilder("scope=");
         sb.append(scope).append(",");
         sb.append("elementType=").append(elementType).append(",");
-        sb.append("lastPushVersion=").append(lastPushVersions);
+        sb.append("pushVersion=").append(lastPushContexts);
         return sb.toString();
     }
 
-    /**
-     * Getter method for property <tt>lastPushVersions</tt>.
-     *
-     * @return property value of lastPushVersions
-     */
-    public Map<String, Long> getLastPushVersions() {
-        return lastPushVersions;
-    }
-
-    /**
-     * Setter method for property <tt>lastPushVersions </tt>.
-     *
-     * @param lastPushVersions  value to be assigned to property lastPushVersions
-     */
-    public void setLastPushVersions(Map<String, Long> lastPushVersions) {
-        this.lastPushVersions = lastPushVersions;
-    }
-
     /**
      * @see Object#toString()
      */
@@ -156,9 +171,57 @@ public String toString() {
         final StringBuilder sb = new StringBuilder("Subscriber{");
         sb.append("scope=").append(scope);
         sb.append(", elementType=").append(elementType);
-        sb.append(", lastPushVersions=").append(lastPushVersions);
+        sb.append(", lastPushContexts=").append(lastPushContexts);
         sb.append(", super=").append(super.toString());
         sb.append('}');
         return sb.toString();
     }
+
+    /**
+     * change subscriber word cache
+     * @param subscriber
+     * @return
+     */
+    public static Subscriber internSubscriber(Subscriber subscriber) {
+        subscriber.setRegisterId(subscriber.getRegisterId());
+        subscriber.setDataInfoId(subscriber.getDataInfoId());
+        subscriber.setInstanceId(subscriber.getInstanceId());
+        subscriber.setGroup(subscriber.getGroup());
+        subscriber.setDataId(subscriber.getDataId());
+        subscriber.setClientId(subscriber.getClientId());
+        subscriber.setCell(subscriber.getCell());
+        subscriber.setProcessId(subscriber.getProcessId());
+        subscriber.setAppName(subscriber.getAppName());
+
+        return subscriber;
+    }
+
+    static class PushContext {
+        /**
+         * last pushed dataInfo version
+         */
+        private Long pushVersion;
+
+        /**
+         * push pushed dataInfo pubCount
+         */
+        private int  pushPubCount;
+
+        public PushContext(Long pushVersion, int pushPubCount) {
+            this.pushVersion = pushVersion;
+            this.pushPubCount = pushPubCount;
+        }
+
+        /**
+         * @see Object#toString()
+         */
+        @Override
+        public String toString() {
+            final StringBuilder sb = new StringBuilder("PushContext{");
+            sb.append("pushVersion=").append(pushVersion);
+            sb.append(", pushPubCount=").append(pushPubCount);
+            sb.append('}');
+            return sb.toString();
+        }
+    }
 }

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/Watcher.java

@@ -27,4 +27,23 @@ public class Watcher extends BaseInfo {
     public DataType getDataType() {
         return DataType.WATCHER;
     }
+
+    /**
+     * change watcher word cache
+     * @param watcher
+     * @return
+     */
+    public static Watcher internWatcher(Watcher watcher) {
+        watcher.setRegisterId(watcher.getRegisterId());
+        watcher.setDataInfoId(watcher.getDataInfoId());
+        watcher.setInstanceId(watcher.getInstanceId());
+        watcher.setGroup(watcher.getGroup());
+        watcher.setDataId(watcher.getDataId());
+        watcher.setClientId(watcher.getClientId());
+        watcher.setCell(watcher.getCell());
+        watcher.setProcessId(watcher.getProcessId());
+        watcher.setAppName(watcher.getAppName());
+
+        return watcher;
+    }
 }

server/common/model/src/main/java/com/alipay/sofa/registry/common/model/store/WordCache.java

@@ -16,7 +16,8 @@
  */
 package com.alipay.sofa.registry.common.model.store;
 
-import java.util.concurrent.ConcurrentHashMap;
+import com.google.common.collect.Interner;
+import com.google.common.collect.Interners;
 
 /**
  *
@@ -45,7 +46,7 @@ public static WordCache getInstance() {
     /**
      * word cache map
      */
-    private ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
+    private Interner<String> interners = Interners.newWeakInterner();
 
     /**
      *
@@ -56,8 +57,7 @@ public String getWordCache(String s) {
         if (s == null) {
             return null;
         }
-        String oldValue = map.putIfAbsent(s, s);
-        return oldValue == null ? s : oldValue;
+        return interners.intern(s);
     }
 
 }