commit f06800c02ea5045de21c115dee9c68a5a5759434
author StevenLuMT <42990025+StevenLuMT@users.noreply.github.com> Sun Jul 31 12:07:21 2022 +0800
committer GitHub <noreply@github.com> Sun Jul 31 12:07:21 2022 +0800
tree f2f4b6bf985463ac4fbff3818c5dc71f120f9942
parent 67208fb74181faa640e793cd5757712fd9b5d9d5

indexDir compatible fix for issue #3430 (#3433)

Descriptions of the changes in this PR:

### Motivation
Fixes https://github.com/apache/bookkeeper/issues/3430, make cookie validation is compatible with old version.
I think we should check indexDirs when configuration‘s indexDirectories is setted  rather than avoiding the check 

### Changes
1. add compatible check 
2. add testcase to cover the new code

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

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
index fc73c42..660668f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
+++ b/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 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, the indexDirs's default value is 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

diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieIndexDirTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieIndexDirTest.java
index 71c160b..3caf076 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieIndexDirTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieIndexDirTest.java
@@ -971,4 +971,34 @@
         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());
+        }
+    }
 }