Repository: apache/shardingsphere
Issue: When the application restarts，LEAF_SEGMENT Distributed Primary Key Generation Not working properly in RC2

## Bug Report

### Which version of ShardingSphere did you use?
sharding-orchestration:4.0.0-RC2
### Which project did you use? Sharding-JDBC or Sharding-Proxy?
Sharding-JDBC 
### Expected behavior
When the program restarts, the primary key of LEAF_SEGMENT distribution can be generated normally.
### Actual behavior
When the program restarts, the program throws a null pointer exception,

### Steps to reproduce the behavior, such as: SQL to execute, sharding rule configuration, when exception occur etc.

Caused by: java.lang.NullPointerException
	at org.apache.shardingsphere.orchestration.reg.zookeeper.curator.CuratorZookeeperRegistryCenter.tryLock(CuratorZookeeperRegistryCenter.java:306)
	at org.apache.shardingsphere.orchestration.internal.keygen.LeafSegmentKeyGenerator.incrementCacheId(LeafSegmentKeyGenerator.java:136)

### Reason analyze (If you can)
When the program restarts, **CuratorZookeeperRegistryCenter leafLock is not initialized properly，**
related code:
1. [LeafSegmentKeyGenerator.java#L95](https://github.com/apache/incubator-shardingsphere/blob/dev/sharding-orchestration/sharding-orchestration-core/src/main/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGenerator.java#L95) 
2. [CuratorZookeeperRegistryCenter.java:#306](https://github.com/apache/incubator-shardingsphere/blob/dev/sharding-orchestration/sharding-orchestration-reg/sharding-orchestration-reg-zookeeper-curator/src/main/java/org/apache/shardingsphere/orchestration/reg/zookeeper/curator/CuratorZookeeperRegistryCenter.java#L306)


Changed files:
sharding-orchestration/sharding-orchestration-zookeeper-curator-integration-test/src/test/java/org/apache/shardingsphere/orchestration/zookeeper/curator/integration/test/LeafSegmentKeyGeneratorIT.java
sharding-orchestration/sharding-orchestration-core/src/main/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGenerator.java
sharding-orchestration/sharding-orchestration-core/src/test/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGeneratorTest.java

diff --git a/sharding-orchestration/sharding-orchestration-core/src/main/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGenerator.java b/sharding-orchestration/sharding-orchestration-core/src/main/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGenerator.java
index c0257790749..8352d22f5ba 100644
--- a/sharding-orchestration/sharding-orchestration-core/src/main/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGenerator.java
+++ b/sharding-orchestration/sharding-orchestration-core/src/main/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGenerator.java
@@ -38,119 +38,125 @@ import java.util.concurrent.SynchronousQueue;
  * @author wangguangyuan
  */
 public final class LeafSegmentKeyGenerator implements ShardingKeyGenerator {
-
+    
     private static final String DEFAULT_NAMESPACE = "leaf_segment";
-
+    
     private static final String DEFAULT_STEP = "10000";
-
+    
     private static final String DEFAULT_INITIAL_VALUE = "1";
-
+    
     private static final String DEFAULT_REGISTRY_CENTER = "zookeeper";
-
-    private static final float DEFAULT_THRESHOLD = 0.5F;
-
+    
     private static final String SLANTING_BAR = "/";
-
+    
     private static final String REGULAR_PATTERN = "^((?!/).)*$";
-
+    
     private final ExecutorService incrementCacheIdExecutor;
-
+    
     private final SynchronousQueue<Long> cacheIdQueue;
-
+    
     private RegistryCenter leafRegistryCenter;
-
+    
     private long id;
-
+    
     private long step;
-
+    
     @Getter
     @Setter
     private Properties properties = new Properties();
-
+    
     public LeafSegmentKeyGenerator() {
         incrementCacheIdExecutor = Executors.newSingleThreadExecutor();
         cacheIdQueue = new SynchronousQueue<>();
     }
-
+    
     @Override
     public String getType() {
         return "LEAF_SEGMENT";
     }
-
+    
     @Override
     public synchronized Comparable<?> generateKey() {
         String leafKey = getLeafKey();
         if (null == leafRegistryCenter) {
             initLeafSegmentKeyGenerator(leafKey);
-            return id;
+        }else{
+            increaseIdWhenLeafKeyStoredInCenter(leafKey);       
         }
-        increaseIdWhenLeafKeyStoredInCenter(leafKey);
         return id;
     }
-
+    
     private void initLeafSegmentKeyGenerator(final String leafKey) {
         RegistryCenterConfiguration leafConfiguration = getRegistryCenterConfiguration();
         leafRegistryCenter = new RegistryCenterServiceLoader().load(leafConfiguration);
-        if (leafRegistryCenter.isExisted(leafKey)) {
-            id = incrementCacheId(leafKey, getStep());
-        } else {
-            id = getInitialValue();
-            leafRegistryCenter.persist(leafKey, String.valueOf(id));
-            leafRegistryCenter.initLock(leafKey);
-        }
         step = getStep();
+        id = initializeId(leafKey);
+        initializeLeafKeyInCenter(leafKey, id, step);
+        initializeCacheIdAsynchronous(id, step);    
     }
-
+    
     private void increaseIdWhenLeafKeyStoredInCenter(final String leafKey) {
+        if ((id % step) == 0) {
+            id = tryTakeCacheId();
+            incrementCacheIdAsynchronous(leafKey, step - (id % step));
+        }
         ++id;
-        if (((id % step) >= (step * DEFAULT_THRESHOLD - 1)) && cacheIdQueue.isEmpty()) {
-            incrementCacheIdAsynchronous(leafKey, step);
+    }
+    
+    private long initializeId(final String leafKey){
+        if (leafRegistryCenter.isExisted(leafKey)) {
+            return Long.parseLong(leafRegistryCenter.getDirectly(leafKey))+1;
+        } else {
+            return getInitialValue();
         }
-        if ((id % step) == (step - 1)) {
-            id = tryTakeCacheId();
+    }
+    
+    private void initializeLeafKeyInCenter(final String leafKey, final long id, final long step){
+        leafRegistryCenter.initLock(leafKey);
+        while(!leafRegistryCenter.tryLock()){
         }
+        leafRegistryCenter.persist(leafKey, String.valueOf(id + step - id % step));
+        leafRegistryCenter.tryRelease();
     }
 
-    private RegistryCenterConfiguration getRegistryCenterConfiguration() {
-        RegistryCenterConfiguration result = new RegistryCenterConfiguration(getRegistryCenterType(), properties);
-        result.setNamespace(DEFAULT_NAMESPACE);
-        result.setServerLists(getServerList());
-        result.setDigest(getDigest());
-        return result;
-    }
+    private void initializeCacheIdAsynchronous(final long id, final long step){
+        incrementCacheIdExecutor.execute(new Runnable() {
 
+            @Override
+            public void run() {
+                tryPutCacheId(id + step - id % step);
+            }
+        });
+    }
+    
     private void incrementCacheIdAsynchronous(final String leafKey, final long step) {
         incrementCacheIdExecutor.execute(new Runnable() {
 
             @Override
             public void run() {
-                long id = incrementCacheId(leafKey, step);
-                tryPutCacheId(id);
+                long newId = incrementCacheId(leafKey, step);
+                tryPutCacheId(newId);
             }
         });
     }
-
+    
+    private RegistryCenterConfiguration getRegistryCenterConfiguration() {
+        RegistryCenterConfiguration result = new RegistryCenterConfiguration(getRegistryCenterType(), properties);
+        result.setNamespace(DEFAULT_NAMESPACE);
+        result.setServerLists(getServerList());
+        result.setDigest(getDigest());
+        return result;
+    }
+    
     @SneakyThrows
     private long incrementCacheId(final String leafKey, final long step) {
-        long result = Long.MIN_VALUE;
-        boolean lockIsAcquired = leafRegistryCenter.tryLock();
-        if (lockIsAcquired) {
-            result = updateCacheIdInCenter(leafKey, step);
-            leafRegistryCenter.tryRelease();
+        while(!leafRegistryCenter.tryLock()){
         }
+        long result = updateCacheIdInCenter(leafKey, step);
+        leafRegistryCenter.tryRelease();
         return result;
     }
-
-    @SneakyThrows
-    private void tryPutCacheId(final long id) {
-        cacheIdQueue.put(id);
-    }
-
-    @SneakyThrows
-    private long tryTakeCacheId() {
-        return cacheIdQueue.take();
-    }
-
+    
     private long updateCacheIdInCenter(final String leafKey, final long step) {
         String cacheIdInString = leafRegistryCenter.getDirectly(leafKey);
         if (Strings.isNullOrEmpty(cacheIdInString)) {
@@ -161,36 +167,46 @@ public final class LeafSegmentKeyGenerator implements ShardingKeyGenerator {
         leafRegistryCenter.update(leafKey, String.valueOf(result));
         return result;
     }
-
+    
+    @SneakyThrows
+    private void tryPutCacheId(final long id) {
+        cacheIdQueue.put(id);
+    }
+    
+    @SneakyThrows
+    private long tryTakeCacheId() {
+        return cacheIdQueue.take();
+    }
+    
     private long getStep() {
         long result = Long.parseLong(properties.getProperty("step", DEFAULT_STEP));
         Preconditions.checkArgument(result > 0L && result < Long.MAX_VALUE);
         return result;
     }
-
+    
     private long getInitialValue() {
         long result = Long.parseLong(properties.getProperty("initialValue", DEFAULT_INITIAL_VALUE));
         Preconditions.checkArgument(result >= 0L && result < Long.MAX_VALUE);
         return result;
     }
-
+    
     private String getLeafKey() {
         String leafKey = properties.getProperty("leafKey");
         Preconditions.checkArgument(!Strings.isNullOrEmpty(leafKey));
         Preconditions.checkArgument(leafKey.matches(REGULAR_PATTERN));
         return SLANTING_BAR + leafKey;
     }
-
+    
     private String getServerList() {
         String result = properties.getProperty("serverList");
         Preconditions.checkArgument(!Strings.isNullOrEmpty(result));
         return result;
     }
-
+    
     private String getDigest() {
         return properties.getProperty("digest");
     }
-
+    
     private String getRegistryCenterType() {
         return properties.getProperty("registryCenterType", DEFAULT_REGISTRY_CENTER);
     }
diff --git a/sharding-orchestration/sharding-orchestration-core/src/test/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGeneratorTest.java b/sharding-orchestration/sharding-orchestration-core/src/test/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGeneratorTest.java
index 789248c9ee9..fafca66265c 100644
--- a/sharding-orchestration/sharding-orchestration-core/src/test/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGeneratorTest.java
+++ b/sharding-orchestration/sharding-orchestration-core/src/test/java/org/apache/shardingsphere/orchestration/internal/keygen/LeafSegmentKeyGeneratorTest.java
@@ -17,7 +17,9 @@
 
 package org.apache.shardingsphere.orchestration.internal.keygen;
 
+import org.junit.FixMethodOrder;
 import org.junit.Test;
+import org.junit.runners.MethodSorters;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -32,6 +34,7 @@ import java.util.concurrent.Executors;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertThat;
 
+@FixMethodOrder(value = MethodSorters.NAME_ASCENDING)
 public final class LeafSegmentKeyGeneratorTest {
     
     private final LeafSegmentKeyGenerator leafSegmentKeyGenerator = new LeafSegmentKeyGenerator();
@@ -54,7 +57,7 @@ public final class LeafSegmentKeyGeneratorTest {
         Properties properties = new Properties();
         properties.setProperty("serverList", "127.0.0.1:2181");
         properties.setProperty("initialValue", "100001");
-        properties.setProperty("step", "3");
+        properties.setProperty("step", "5");
         properties.setProperty("digest", "");
         properties.setProperty("leafKey", "test_table_1");
         properties.setProperty("registryCenterType", "ThirdTestRegistryCenter");
@@ -67,6 +70,42 @@ public final class LeafSegmentKeyGeneratorTest {
         assertThat(actual, is(expected));
     }
     
+    @Test
+    public void assertGenerateKeyWithFirstSpecialStep() {
+        Properties properties = new Properties();
+        properties.setProperty("serverList", "127.0.0.1:2181");
+        properties.setProperty("initialValue", "100001");
+        properties.setProperty("step", "3");
+        properties.setProperty("digest", "");
+        properties.setProperty("leafKey", "test_table_6");
+        properties.setProperty("registryCenterType", "ThirdTestRegistryCenter");
+        leafSegmentKeyGenerator.setProperties(properties);
+        List<Comparable<?>> expected = Arrays.<Comparable<?>>asList(100001L, 100002L, 100003L, 100004L, 100005L, 100006L, 100007L, 100008L, 100009L, 100010L);
+        List<Comparable<?>> actual = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            actual.add(leafSegmentKeyGenerator.generateKey());
+        }
+        assertThat(actual, is(expected));
+    }
+    
+    @Test
+    public void assertGenerateKeyWithSecondSpecialStep() {
+        Properties properties = new Properties();
+        properties.setProperty("serverList", "127.0.0.1:2181");
+        properties.setProperty("initialValue", "100001");
+        properties.setProperty("step", "7");
+        properties.setProperty("digest", "");
+        properties.setProperty("leafKey", "test_table_7");
+        properties.setProperty("registryCenterType", "ThirdTestRegistryCenter");
+        leafSegmentKeyGenerator.setProperties(properties);
+        List<Comparable<?>> expected = Arrays.<Comparable<?>>asList(100001L, 100002L, 100003L, 100004L, 100005L, 100006L, 100007L, 100008L, 100009L, 100010L);
+        List<Comparable<?>> actual = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            actual.add(leafSegmentKeyGenerator.generateKey());
+        }
+        assertThat(actual, is(expected));
+    }
+    
     @Test
     public void assertGenerateKeyWithMultipleThreads() throws Exception {
         int threadNumber = Runtime.getRuntime().availableProcessors() << 1;
@@ -127,7 +166,7 @@ public final class LeafSegmentKeyGeneratorTest {
         properties.setProperty("serverList", "127.0.0.1:2181");
         properties.setProperty("initialValue", "100001");
         properties.setProperty("digest", "");
-        properties.setProperty("leafKey", "test_table_6");
+        properties.setProperty("leafKey", "test_table_4");
         properties.setProperty("registryCenterType", "ThirdTestRegistryCenter");
         leafSegmentKeyGenerator.setProperties(properties);
         Set<Comparable<?>> actual = new HashSet<>();
@@ -152,7 +191,7 @@ public final class LeafSegmentKeyGeneratorTest {
         properties.setProperty("serverList", "127.0.0.1:2181");
         properties.setProperty("step", "3");
         properties.setProperty("digest", "");
-        properties.setProperty("leafKey", "test_table_7");
+        properties.setProperty("leafKey", "test_table_5");
         properties.setProperty("registryCenterType", "ThirdTestRegistryCenter");
         leafSegmentKeyGenerator.setProperties(properties);
         int taskNumber = threadNumber << 2;
diff --git a/sharding-orchestration/sharding-orchestration-zookeeper-curator-integration-test/src/test/java/org/apache/shardingsphere/orchestration/zookeeper/curator/integration/test/LeafSegmentKeyGeneratorIT.java b/sharding-orchestration/sharding-orchestration-zookeeper-curator-integration-test/src/test/java/org/apache/shardingsphere/orchestration/zookeeper/curator/integration/test/LeafSegmentKeyGeneratorIT.java
index 6ce4c1023c5..5186d979c3f 100644
--- a/sharding-orchestration/sharding-orchestration-zookeeper-curator-integration-test/src/test/java/org/apache/shardingsphere/orchestration/zookeeper/curator/integration/test/LeafSegmentKeyGeneratorIT.java
+++ b/sharding-orchestration/sharding-orchestration-zookeeper-curator-integration-test/src/test/java/org/apache/shardingsphere/orchestration/zookeeper/curator/integration/test/LeafSegmentKeyGeneratorIT.java
@@ -20,7 +20,9 @@ package org.apache.shardingsphere.orchestration.zookeeper.curator.integration.te
 import org.apache.shardingsphere.orchestration.internal.keygen.LeafSegmentKeyGenerator;
 import org.apache.shardingsphere.orchestration.zookeeper.curator.integration.util.EmbedTestingServer;
 import org.junit.BeforeClass;
+import org.junit.FixMethodOrder;
 import org.junit.Test;
+import org.junit.runners.MethodSorters;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -40,6 +42,7 @@ import static org.junit.Assert.assertThat;
  *
  * @author wangguangyuan
  */
+@FixMethodOrder(value = MethodSorters.NAME_ASCENDING)
 public final class LeafSegmentKeyGeneratorIT {
     
     private final LeafSegmentKeyGenerator leafSegmentKeyGenerator = new LeafSegmentKeyGenerator();
@@ -67,6 +70,78 @@ public final class LeafSegmentKeyGeneratorIT {
         assertThat(actual, is(expected));
     }
     
+    @Test
+    public void assertGenerateKeyWithFirstSpecialStep() {
+        Properties properties = new Properties();
+        properties.setProperty("serverList", "127.0.0.1:3181");
+        properties.setProperty("initialValue", "100001");
+        properties.setProperty("step", "5");
+        properties.setProperty("digest", "");
+        properties.setProperty("leafKey", "test_table_6");
+        properties.setProperty("registryCenterType", "zookeeper");
+        leafSegmentKeyGenerator.setProperties(properties);
+        List<Comparable<?>> expected = Arrays.<Comparable<?>>asList(100001L, 100002L, 100003L, 100004L, 100005L, 100006L, 100007L, 100008L, 100009L, 100010L);
+        List<Comparable<?>> actual = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            actual.add(leafSegmentKeyGenerator.generateKey());
+        }
+        assertThat(actual, is(expected));
+    }
+    
+    @Test
+    public void assertGenerateKeyWithFirstSpecialStepAgain() {
+        Properties properties = new Properties();
+        properties.setProperty("serverList", "127.0.0.1:3181");
+        properties.setProperty("initialValue", "100001");
+        properties.setProperty("step", "5");
+        properties.setProperty("digest", "");
+        properties.setProperty("leafKey", "test_table_6");
+        properties.setProperty("registryCenterType", "zookeeper");
+        leafSegmentKeyGenerator.setProperties(properties);
+        List<Comparable<?>> expected = Arrays.<Comparable<?>>asList(100011L, 100012L, 100013L, 100014L, 100015L, 100016L, 100017L, 100018L, 100019L, 100020L);
+        List<Comparable<?>> actual = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            actual.add(leafSegmentKeyGenerator.generateKey());
+        }
+        assertThat(actual, is(expected));
+    }
+    
+    @Test
+    public void assertGenerateKeyWithSecondSpecialStep() {
+        Properties properties = new Properties();
+        properties.setProperty("serverList", "127.0.0.1:3181");
+        properties.setProperty("initialValue", "100001");
+        properties.setProperty("step", "7");
+        properties.setProperty("digest", "");
+        properties.setProperty("leafKey", "test_table_7");
+        properties.setProperty("registryCenterType", "zookeeper");
+        leafSegmentKeyGenerator.setProperties(properties);
+        List<Comparable<?>> expected = Arrays.<Comparable<?>>asList(100001L, 100002L, 100003L, 100004L, 100005L, 100006L, 100007L, 100008L, 100009L, 100010L);
+        List<Comparable<?>> actual = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            actual.add(leafSegmentKeyGenerator.generateKey());
+        }
+        assertThat(actual, is(expected));
+    }
+    
+    @Test
+    public void assertGenerateKeyWithSecondSpecialStepAgain() {
+        Properties properties = new Properties();
+        properties.setProperty("serverList", "127.0.0.1:3181");
+        properties.setProperty("initialValue", "100001");
+        properties.setProperty("step", "7");
+        properties.setProperty("digest", "");
+        properties.setProperty("leafKey", "test_table_7");
+        properties.setProperty("registryCenterType", "zookeeper");
+        leafSegmentKeyGenerator.setProperties(properties);
+        List<Comparable<?>> expected = Arrays.<Comparable<?>>asList(100017L, 100018L, 100019L, 100020L, 100021L, 100022L, 100023L, 100024L, 100025L, 100026L);
+        List<Comparable<?>> actual = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            actual.add(leafSegmentKeyGenerator.generateKey());
+        }
+        assertThat(actual, is(expected));
+    }
+    
     @Test
     public void assertGenerateKeyWithMultipleThreads() throws Exception {
         int threadNumber = Runtime.getRuntime().availableProcessors() << 1;
@@ -127,7 +202,7 @@ public final class LeafSegmentKeyGeneratorIT {
         properties.setProperty("serverList", "127.0.0.1:3181");
         properties.setProperty("initialValue", "100001");
         properties.setProperty("digest", "");
-        properties.setProperty("leafKey", "test_table_6");
+        properties.setProperty("leafKey", "test_table_4");
         properties.setProperty("registryCenterType", "zookeeper");
         leafSegmentKeyGenerator.setProperties(properties);
         Set<Comparable<?>> actual = new HashSet<>();
@@ -152,7 +227,7 @@ public final class LeafSegmentKeyGeneratorIT {
         properties.setProperty("serverList", "127.0.0.1:3181");
         properties.setProperty("step", "3");
         properties.setProperty("digest", "");
-        properties.setProperty("leafKey", "test_table_7");
+        properties.setProperty("leafKey", "test_table_5");
         properties.setProperty("registryCenterType", "zookeeper");
         leafSegmentKeyGenerator.setProperties(properties);
         int taskNumber = threadNumber << 2;