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

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

@@ -970,4 +970,35 @@ 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("fail to start,error info:");
+            ice.printStackTrace();
+        }
+    }
 }