Make after version5 cookie index dir compatible with before version5.

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

@@ -50,7 +50,6 @@
 import org.apache.bookkeeper.versioning.LongVersion;
 import org.apache.bookkeeper.versioning.Version;
 import org.apache.bookkeeper.versioning.Versioned;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -138,14 +137,20 @@ private boolean verifyLedgerDirs(Cookie c, boolean checkIfSuperSet) {
     }
 
     private boolean verifyIndexDirs(Cookie c, boolean checkIfSuperSet) {
-        // compatible logic:  existed node's cookie has no indexDirs, the indexDirs's default value is ledgerDirs.
-        String indexDirsInConfig = StringUtils.isNotBlank(indexDirs) ? indexDirs : ledgerDirs;
-        String indexDirsInCookie = StringUtils.isNotBlank(c.indexDirs) ? c.indexDirs : c.ledgerDirs;
-
+        // compatible logic:  version 5 introduce `indexDirs`, we shouldn't check `indexDirs` with the before version.
+        if (this.layoutVersion >= 5 && c.layoutVersion < 5) {
+            return true;
+        }
+        if (indexDirs == null && c.indexDirs == null) {
+            return true;
+        }
+        if (indexDirs == null || c.indexDirs == null) {
+            return false;
+        }
         if (!checkIfSuperSet) {
-            return indexDirsInConfig.equals(indexDirsInCookie);
+            return indexDirs.equals(c.indexDirs);
         } else {
-            return isSuperSet(decodeDirPathFromCookie(indexDirsInConfig), decodeDirPathFromCookie(indexDirsInCookie));
+            return isSuperSet(decodeDirPathFromCookie(indexDirs), decodeDirPathFromCookie(c.indexDirs));
         }
     }
 
@@ -194,7 +199,7 @@ public String toString() {
         }
 
         StringBuilder b = new StringBuilder();
-        b.append(CURRENT_COOKIE_LAYOUT_VERSION).append("\n");
+        b.append(layoutVersion).append("\n");
         b.append(builder.build());
         return b.toString();
     }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.bookkeeper.bookie;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import java.io.File;
 import java.io.FileNotFoundException;
@@ -221,7 +222,8 @@ private static void verifyDirsForNewEnvironment(List<File> missedCookieDirs)
         }
     }
 
-    private static void stampNewCookie(ServerConfiguration conf,
+    @VisibleForTesting
+    public static void stampNewCookie(ServerConfiguration conf,
                                        Cookie masterCookie,
                                        RegistrationManager rm,
                                        Version version,

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

@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Random;
 import java.util.Set;
+import java.util.UUID;
 import org.apache.bookkeeper.bookie.BookieException.InvalidCookieException;
 import org.apache.bookkeeper.client.BookKeeperAdmin;
 import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -973,32 +974,69 @@ public void testBookieIdSetting() throws Exception {
     }
 
     /**
-     * Compatibility test
-     * 1. First create bookie without indexDirName
-     * 2. Configure indexDirName to start bookie
+     * Compatibility test.
      */
     @Test
-    public void testNewBookieStartingWithOldCookie() throws Exception {
+    public void testCookieCompatibility() throws Exception {
         String journalDir = newDirectory();
         String[] ledgerDirs = {newDirectory(), newDirectory()};
-        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
-        conf.setJournalDirName(journalDir)
+        String[] indexDirs = {newDirectory()};
+
+        ServerConfiguration confWithDirs = TestBKConfiguration.newServerConfiguration();
+        confWithDirs.setJournalDirName(journalDir)
                 .setLedgerDirNames(ledgerDirs)
+                .setIndexDirName(indexDirs)
                 .setBookiePort(bookiePort)
                 .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
-        validateConfig(conf);
+        String instanceId = UUID.randomUUID().toString();
 
-        conf = TestBKConfiguration.newServerConfiguration();
-        conf.setJournalDirName(journalDir)
+        Cookie.Builder builder = Cookie.generateCookie(confWithDirs);
+        builder.setLayoutVersion(4);
+        builder.setIndexDirs(null);
+        builder.setInstanceId(instanceId);
+        Cookie version4Cookie = builder.build();
+
+        Cookie.Builder builder1 = Cookie.generateCookie(confWithDirs);
+        builder1.setInstanceId(instanceId);
+        Cookie version5CookieWithIndexDirs = builder1.build();
+
+        ServerConfiguration confWithoutDirs = TestBKConfiguration.newServerConfiguration();
+        confWithoutDirs.setJournalDirName(journalDir)
                 .setLedgerDirNames(ledgerDirs)
-                .setIndexDirName(ledgerDirs)
                 .setBookiePort(bookiePort)
                 .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
+        Cookie.Builder builder2 = Cookie.generateCookie(confWithoutDirs);
+        builder2.setInstanceId(instanceId);
+        Cookie version5CookieWithoutIndexDirs = builder2.build();
+
+        version5CookieWithIndexDirs.verify(version4Cookie);
+        version5CookieWithIndexDirs.verifyIsSuperSet(version4Cookie);
+
+        version5CookieWithoutIndexDirs.verify(version4Cookie);
+        version5CookieWithoutIndexDirs.verifyIsSuperSet(version4Cookie);
+
         try {
-            validateConfig(conf);
-        } catch (InvalidCookieException ice) {
-            // error behaviour
-            fail("Validate failed, error info: " + ice.getMessage());
+            version5CookieWithIndexDirs.verify(version5CookieWithoutIndexDirs);
+            fail("Shouldn't compatible");
+        } catch (Exception ignore) {
+        }
+
+        try {
+            version5CookieWithIndexDirs.verifyIsSuperSet(version5CookieWithoutIndexDirs);
+            fail("Shouldn't compatible");
+        } catch (Exception ignore) {
+        }
+
+        try {
+            version5CookieWithoutIndexDirs.verify(version5CookieWithIndexDirs);
+            fail("Shouldn't compatible");
+        } catch (Exception ignore) {
+        }
+
+        try {
+            version5CookieWithoutIndexDirs.verifyIsSuperSet(version5CookieWithIndexDirs);
+            fail("Shouldn't compatible");
+        } catch (Exception ignore) {
         }
     }
 }