indexDir compatible fix for issue #3430

Author: StevenLuMT

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

@@ -50,6 +50,7 @@
 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;
 
@@ -137,18 +138,14 @@ private boolean verifyLedgerDirs(Cookie c, boolean checkIfSuperSet) {
     }
 
     private boolean verifyIndexDirs(Cookie c, boolean checkIfSuperSet) {
-        // Compatible with old cookie
-        if (null == indexDirs && null == c.indexDirs) {
-            return true;
-        }
-        if (null == indexDirs || null == c.indexDirs) {
-            return false;
-        }
+        // compatible logic:  existed node's cookie has no indexDirs, but indexDirs's default value equals ledgerDirs
+        String indexDirsInConfig = StringUtils.isNotBlank(indexDirs) ? indexDirs : ledgerDirs;
+        String indexDirsInCookie = StringUtils.isNotBlank(c.indexDirs) ? c.indexDirs : c.ledgerDirs;
 
         if (!checkIfSuperSet) {
-            return indexDirs.equals(c.indexDirs);
+            return indexDirsInConfig.equals(indexDirsInCookie);
         } else {
-            return isSuperSet(decodeDirPathFromCookie(indexDirs), decodeDirPathFromCookie(c.indexDirs));
+            return isSuperSet(decodeDirPathFromCookie(indexDirsInConfig), decodeDirPathFromCookie(indexDirsInCookie));
         }
     }
 

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

@@ -970,4 +970,34 @@ public void testBookieIdSetting() throws Exception {
         cookie.writeToRegistrationManager(rm, conf, version1);
         Assert.assertTrue(cookie.toString().contains(customBookieId));
     }
+
+    /**
+     * Compatibility test
+     * 1. First create bookie without indexDirName
+     * 2. Configure indexDirName to start bookie
+     */
+    @Test
+    public void testNewBookieStartingWithOldCookie() throws Exception {
+        String journalDir = newDirectory();
+        String[] ledgerDirs = {newDirectory(), newDirectory()};
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        conf.setJournalDirName(journalDir)
+                .setLedgerDirNames(ledgerDirs)
+                .setBookiePort(bookiePort)
+                .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
+        validateConfig(conf);
+
+        conf = TestBKConfiguration.newServerConfiguration();
+        conf.setJournalDirName(journalDir)
+                .setLedgerDirNames(ledgerDirs)
+                .setIndexDirName(ledgerDirs)
+                .setBookiePort(bookiePort)
+                .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
+        try {
+            validateConfig(conf);
+        } catch (InvalidCookieException ice) {
+            // error behaviour
+            fail("Validate failed, error info: " + ice.getMessage());
+        }
+    }
 }