Fix cookie compatible problem.

bookkeeper-proto/src/main/proto/DataFormats.proto

@@ -83,6 +83,7 @@ message CookieFormat {
     required string ledgerDirs = 3;
     optional string instanceId = 4;
     optional string indexDirs = 5;
+    optional bool isNew = 6;
 }
 
 /**

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java

@@ -75,16 +75,18 @@ public class Cookie {
     private final String ledgerDirs;
     private final String indexDirs;
     private final String instanceId;
+    private final boolean isNew;
     private static final String SEPARATOR = "\t";
 
     private Cookie(int layoutVersion, String bookieId, String journalDirs, String ledgerDirs, String instanceId,
-                   String indexDirs) {
+                   String indexDirs, boolean isNew) {
         this.layoutVersion = layoutVersion;
         this.bookieId = bookieId;
         this.journalDirs = journalDirs;
         this.ledgerDirs = ledgerDirs;
         this.instanceId = instanceId;
         this.indexDirs = indexDirs;
+        this.isNew = isNew;
     }
 
     public static String encodeDirPaths(String[] dirs) {
@@ -137,14 +139,19 @@ private boolean verifyLedgerDirs(Cookie c, boolean checkIfSuperSet) {
     }
 
     private boolean verifyIndexDirs(Cookie c, boolean checkIfSuperSet) {
-        // Compatible with old cookie
+        // Compatible with old cookie, the old cookie in rm didn't set indexDirs.
+        if (!c.isNew) {
+            return true;
+        }
         if (null == indexDirs && null == c.indexDirs) {
             return true;
         }
-        if (null == indexDirs || null == c.indexDirs) {
+        if (null == indexDirs) {
+            return false;
+        }
+        if (null == c.indexDirs) {
             return false;
         }
-
         if (!checkIfSuperSet) {
             return indexDirs.equals(c.indexDirs);
         } else {
@@ -195,6 +202,7 @@ public String toString() {
         if (null != indexDirs) {
             builder.setIndexDirs(indexDirs);
         }
+        builder.setIsNew(isNew);
 
         StringBuilder b = new StringBuilder();
         b.append(CURRENT_COOKIE_LAYOUT_VERSION).append("\n");
@@ -243,6 +251,7 @@ private static Builder parse(BufferedReader reader) throws IOException {
             if (null != data.getIndexDirs() && !data.getIndexDirs().isEmpty()) {
                 cBuilder.setIndexDirs(data.getIndexDirs());
             }
+            cBuilder.setIsNew(data.getIsNew());
         }
         return cBuilder;
     }
@@ -436,6 +445,7 @@ public static class Builder {
         private String ledgerDirs = null;
         private String instanceId = null;
         private String indexDirs = null;
+        private boolean isNew = true;
 
         private Builder() {
         }
@@ -480,8 +490,13 @@ public Builder setIndexDirs(String indexDirs) {
             return this;
         }
 
+        public Builder setIsNew(boolean isNew) {
+            this.isNew = isNew;
+            return this;
+        }
+
         public Cookie build() {
-            return new Cookie(layoutVersion, bookieId, journalDirs, ledgerDirs, instanceId, indexDirs);
+            return new Cookie(layoutVersion, bookieId, journalDirs, ledgerDirs, instanceId, indexDirs, isNew);
         }
     }
 

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieIndexDirTest.java

@@ -107,7 +107,7 @@ private static List<File> currentDirectoryList(File[] dirs) {
         return Arrays.asList(BookieImpl.getCurrentDirectories(dirs));
     }
 
-    private void validateConfig(ServerConfiguration conf) throws Exception {
+    private List<File> getDirs(ServerConfiguration conf) throws Exception {
         List<File> dirs = new ArrayList<>();
         for (File f : conf.getJournalDirs()) {
             File cur = BookieImpl.getCurrentDirectory(f);
@@ -126,9 +126,13 @@ private void validateConfig(ServerConfiguration conf) throws Exception {
                 BookieImpl.checkDirectoryStructure(cur);
             }
         }
+        return dirs;
+    }
+
+    private void validateConfig(ServerConfiguration conf) throws Exception {
+        List<File> dirs = getDirs(conf);
         LegacyCookieValidation cookieValidation = new LegacyCookieValidation(conf, rm);
         cookieValidation.checkCookies(dirs);
-
     }
 
     /**
@@ -970,4 +974,72 @@ public void testBookieIdSetting() throws Exception {
         cookie.writeToRegistrationManager(rm, conf, version1);
         Assert.assertTrue(cookie.toString().contains(customBookieId));
     }
+
+    /**
+     * Tests compatible with the old version cookie already exists in RM.
+     */
+    @Test
+    public void testCompatibleWithExistRM() throws Exception {
+        final String customBookieId = "myCustomBookieId" + new Random().nextInt();
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        conf.setJournalDirName(newDirectory())
+                .setLedgerDirNames(new String[] { newDirectory() , newDirectory() })
+                .setIndexDirName(new String[] { newDirectory() , newDirectory() })
+                .setBookiePort(bookiePort)
+                .setBookieId(customBookieId)
+                .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
+
+        //Old version cookie didn't contains 'isNew' and 'indexDirs'
+        Cookie.Builder builder = Cookie.generateCookie(conf);
+        builder.setIsNew(false);
+        builder.setIndexDirs(null);
+        String instanceId = rm.getClusterInstanceId();
+        if (instanceId != null) {
+            builder.setInstanceId(instanceId);
+        }
+        Cookie oldVersionCookie = builder.build();
+        List<File> dirs = getDirs(conf);
+        for (File dir : dirs) {
+            oldVersionCookie.writeToDirectory(dir);
+        }
+        oldVersionCookie.writeToRegistrationManager(rm, conf, Version.NEW);
+
+        validateConfig(conf);
+    }
+
+    /**
+     * Tests compatible with the old version cookie already exists in RM.
+     */
+    @Test
+    public void testCompatibleWithExistRMOnFailed() throws Exception {
+        final String customBookieId = "myCustomBookieId" + new Random().nextInt();
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        conf.setJournalDirName(newDirectory())
+                .setLedgerDirNames(new String[] { newDirectory() , newDirectory() })
+                .setIndexDirName(new String[] { newDirectory() , newDirectory() })
+                .setBookiePort(bookiePort)
+                .setBookieId(customBookieId)
+                .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
+
+        //Build new version cookie without indexDirs
+        Cookie.Builder builder = Cookie.generateCookie(conf);
+        builder.setIsNew(true);
+        builder.setIndexDirs(null);
+        String instanceId = rm.getClusterInstanceId();
+        if (instanceId != null) {
+            builder.setInstanceId(instanceId);
+        }
+        Cookie newVersionCookie = builder.build();
+        List<File> dirs = getDirs(conf);
+        for (File dir : dirs) {
+            newVersionCookie.writeToDirectory(dir);
+        }
+        newVersionCookie.writeToRegistrationManager(rm, conf, Version.NEW);
+
+        try {
+            validateConfig(conf);
+            Assert.fail("Should validate failed.");
+        } catch (BookieException.InvalidCookieException e) {
+        }
+    }
 }