Clean the useless versions after metadata version switched

infra/common/src/main/java/org/apache/shardingsphere/infra/metadata/version/MetaDataVersion.java

@@ -55,9 +55,19 @@ public String getActiveVersionNodePath() {
     /**
      * Get versions node path.
      *
+     * @param version version
      * @return path of versions node
      */
-    public String getVersionsNodePath() {
-        return String.join("/", key, VERSIONS, currentActiveVersion);
+    public String getVersionsNodePath(final String version) {
+        return String.join("/", key, VERSIONS, version);
+    }
+
+    /**
+     * Get versions path.
+     *
+     * @return path of versions
+     */
+    public String getVersionsPath() {
+        return String.join("/", key, VERSIONS);
     }
 }

infra/common/src/test/java/org/apache/shardingsphere/infra/metadata/version/MetaDataVersionTest.java

@@ -31,6 +31,11 @@ void assertGetActiveVersionNodePath() {
 
     @Test
     void assertGetVersionsNodePath() {
-        assertThat(new MetaDataVersion("foo", "0", "1").getVersionsNodePath(), is("foo/versions/0"));
+        assertThat(new MetaDataVersion("foo", "0", "1").getVersionsNodePath("0"), is("foo/versions/0"));
+    }
+
+    @Test
+    void assertGetVersionsPath() {
+        assertThat(new MetaDataVersion("foo", "0", "1").getVersionsPath(), is("foo/versions"));
     }
 }

kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/service/version/MetaDataVersionPersistService.java

@@ -42,7 +42,9 @@ public void switchActiveVersion(final Collection<MetaDataVersion> metaDataVersio
                 continue;
             }
             repository.persist(each.getActiveVersionNodePath(), each.getNextActiveVersion());
-            repository.delete(each.getVersionsNodePath());
+            getVersions(each.getVersionsPath()).stream()
+                    .filter(version -> !version.equals(each.getNextActiveVersion()))
+                    .forEach(version -> repository.delete(each.getVersionsNodePath(version)));
         }
     }
 
@@ -59,7 +61,7 @@ public String getVersionPathByActiveVersion(final String path, final String acti
     @Override
     public List<String> getVersions(final String path) {
         List<String> result = repository.getChildrenKeys(path);
-        if (result.size() > 1) {
+        if (result.size() > 2) {
             log.warn("There are multiple versions of ：{}, please check the configuration.", path);
             result.sort((v1, v2) -> Integer.compare(Integer.parseInt(v2), Integer.parseInt(v1)));
         }

kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/service/version/MetaDataVersionPersistServiceTest.java

@@ -47,6 +47,7 @@ void setUp() {
 
     @Test
     void assertSwitchActiveVersion() {
+        when(repository.getChildrenKeys("foo_db/versions")).thenReturn(Arrays.asList("1", "0"));
         persistService.switchActiveVersion(Arrays.asList(new MetaDataVersion("foo_db", "0", "1"), new MetaDataVersion("bar_db", "2", "2")));
         verify(repository).persist("foo_db/active_version", "1");
         verify(repository).delete("foo_db/versions/0");