geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From n...@apache.org
Subject [geode] branch develop updated: GEODE-5082: Reindex not allowed on mixed server versions. (#1803)
Date Sun, 22 Apr 2018 23:01:35 GMT
This is an automated email from the ASF dual-hosted git repository.

nnag pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git


The following commit(s) were added to refs/heads/develop by this push:
     new 5536886  GEODE-5082: Reindex not allowed on mixed server versions. (#1803)
5536886 is described below

commit 55368864e117e1bd43483e995ec294ffe0377918
Author: Nabarun Nag <nabarunnag@users.noreply.github.com>
AuthorDate: Sun Apr 22 16:01:30 2018 -0700

    GEODE-5082: Reindex not allowed on mixed server versions. (#1803)
    
             * Lucene reindex is not allowed when the members hosting the region are of different
versions.
    	* Any exception during the reindex will now clear the lucene index from the index defined
map.
---
 .../cache/lucene/internal/LuceneServiceImpl.java   |  49 +++++--
 .../LuceneSearchWithRollingUpgradeDUnit.java       | 145 ++++++++++++++++++---
 .../internal/LuceneServiceImplJUnitTest.java       |   5 +
 3 files changed, 168 insertions(+), 31 deletions(-)

diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
index 5756d71..211b508 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
@@ -64,8 +64,10 @@ import org.apache.geode.cache.lucene.internal.results.LuceneGetPageFunction;
 import org.apache.geode.cache.lucene.internal.results.PageResults;
 import org.apache.geode.cache.lucene.internal.xml.LuceneServiceXmlGenerator;
 import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.DSFIDFactory;
 import org.apache.geode.internal.DataSerializableFixedID;
+import org.apache.geode.internal.Version;
 import org.apache.geode.internal.cache.BucketNotFoundException;
 import org.apache.geode.internal.cache.BucketRegion;
 import org.apache.geode.internal.cache.CacheService;
@@ -196,24 +198,45 @@ public class LuceneServiceImpl implements InternalLuceneService {
     // We must always register the index (this is where IndexAlreadyExistsException is detected)
     registerDefinedIndex(indexName, regionPath, new LuceneIndexCreationProfile(indexName,
         regionPath, fields, analyzer, fieldAnalyzers, serializer));
+    try {
+      // If the region does not yet exist, install LuceneRegionListener and return
+      PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionPath);
+      if (region == null) {
+        LuceneRegionListener regionListener = new LuceneRegionListener(this, cache, indexName,
+            regionPath, fields, analyzer, fieldAnalyzers, serializer);
+        cache.addRegionListener(regionListener);
+        return;
+      } else if (allowOnExistingRegion) {
+        validateAllMembersAreTheSameVersion(region);
+      }
 
-    // If the region does not yet exist, install LuceneRegionListener and return
-    PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionPath);
-    if (region == null) {
-      LuceneRegionListener regionListener = new LuceneRegionListener(this, cache, indexName,
-          regionPath, fields, analyzer, fieldAnalyzers, serializer);
-      cache.addRegionListener(regionListener);
-      return;
-    }
+      if (!allowOnExistingRegion) {
+        definedIndexMap.remove(LuceneServiceImpl.getUniqueIndexName(indexName, regionPath));
+        throw new IllegalStateException("The lucene index must be created before region");
+      }
 
-    if (!allowOnExistingRegion) {
+      // do work normally handled by LuceneRegionListener (if region already exists)
+      createIndexOnExistingRegion(region, indexName, regionPath, fields, analyzer, fieldAnalyzers,
+          serializer);
+    } catch (Exception exception) {
       definedIndexMap.remove(LuceneServiceImpl.getUniqueIndexName(indexName, regionPath));
-      throw new IllegalStateException("The lucene index must be created before region");
+      throw exception;
     }
+  }
 
-    // do work normally handled by LuceneRegionListener (if region already exists)
-    createIndexOnExistingRegion(region, indexName, regionPath, fields, analyzer, fieldAnalyzers,
-        serializer);
+  protected void validateAllMembersAreTheSameVersion(PartitionedRegion region) {
+    Set<InternalDistributedMember> remoteMembers = region.getRegionAdvisor().adviseAllPRNodes();
+    Version localVersion =
+        cache.getDistributionManager().getDistributionManagerId().getVersionObject();
+    if (!remoteMembers.isEmpty()) {
+      for (InternalDistributedMember remoteMember : remoteMembers) {
+        if (!remoteMember.getVersionObject().equals(localVersion)) {
+          throw new IllegalStateException(
+              "The lucene index cannot be created on a existing region if all members hosting
the region : "
+                  + region.getFullPath() + ", are not the same Apache Geode version ");
+        }
+      }
+    }
   }
 
   private void createIndexOnExistingRegion(PartitionedRegion region, String indexName,
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
index 0abe412..ee98f24 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
@@ -52,6 +52,7 @@ import org.apache.geode.internal.AvailablePortHelper;
 import org.apache.geode.internal.Version;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
 import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.DistributedTestUtils;
 import org.apache.geode.test.dunit.Host;
 import org.apache.geode.test.dunit.IgnoredException;
@@ -140,6 +141,83 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
     executeLuceneQueryWithServerRollOvers("persistentPartitioned", oldVersion);
   }
 
+  @Test
+  public void luceneReindexShouldBeSuccessfulWhenAllServersRollToCurrentVersion() throws
Exception {
+    final Host host = Host.getHost(0);
+    VM locator1 = host.getVM(oldVersion, 0);
+    VM server1 = host.getVM(oldVersion, 1);
+    VM server2 = host.getVM(oldVersion, 2);
+
+    final String regionName = "aRegion";
+    RegionShortcut shortcut = RegionShortcut.PARTITION_REDUNDANT;
+
+    int locatorPort = AvailablePortHelper.getRandomAvailableTCPPort();
+    DistributedTestUtils.deleteLocatorStateFile(locatorPort);
+
+    String hostName = NetworkUtils.getServerHostName(host);
+    String locatorString = getLocatorString(locatorPort);
+    String regionType = "partitionedRedundant";
+    try {
+      locator1.invoke(
+          invokeStartLocator(hostName, locatorPort, getLocatorPropertiesPre91(locatorString)));
+      invokeRunnableInVMs(invokeCreateCache(getSystemProperties(new int[] {locatorPort})),
server1,
+          server2);
+
+      // Locators before 1.4 handled configuration asynchronously.
+      // We must wait for configuration configuration to be ready, or confirm that it is
disabled.
+      locator1.invoke(
+          () -> Awaitility.await().atMost(65, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS)
+              .until(() -> assertTrue(
+                  !InternalLocator.getLocator().getConfig().getEnableClusterConfiguration()
+                      || InternalLocator.getLocator().isSharedConfigurationRunning())));
+
+      locator1 =
+          rollLocatorToCurrent(locator1, hostName, locatorPort, getTestMethodName(), locatorString);
+
+      server1 = rollServerToCurrentAndCreateRegionOnly(server1, regionType, null, shortcut.name(),
+          regionName, new int[] {locatorPort});
+
+      invokeRunnableInVMs(invokeCreateRegion(regionName, shortcut.name()), server2);
+      try {
+        server1.invoke(() -> createLuceneIndexOnExistingRegion(cache, regionName, INDEX_NAME));
+        fail();
+      } catch (Exception exception) {
+        if (!exception.getCause().getCause().getMessage()
+            .contains("are not the same Apache Geode version")) {
+          exception.printStackTrace();
+          fail();
+        }
+      }
+
+      int expectedRegionSize = 10;
+      putSerializableObject(server1, regionName, 0, expectedRegionSize);
+
+      server2 = rollServerToCurrentAndCreateRegionOnly(server2, regionType, null, shortcut.name(),
+          regionName, new int[] {locatorPort});
+
+
+      AsyncInvocation ai1 = server1
+          .invokeAsync(() -> createLuceneIndexOnExistingRegion(cache, regionName, INDEX_NAME));
+
+      AsyncInvocation ai2 = server2
+          .invokeAsync(() -> createLuceneIndexOnExistingRegion(cache, regionName, INDEX_NAME));
+
+      ai1.join();
+      ai2.join();
+
+      ai1.checkException();
+      ai2.checkException();
+
+      expectedRegionSize += 10;
+      putSerializableObjectAndVerifyLuceneQueryResult(server2, regionName, expectedRegionSize,
15,
+          25, server1, server2);
+
+    } finally {
+      invokeRunnableInVMs(true, invokeStopLocator(), locator1);
+      invokeRunnableInVMs(true, invokeCloseCache(), server1, server2);
+    }
+  }
+
   // 2 locator, 2 servers
   @Test
   public void luceneQueryReturnsCorrectResultAfterTwoLocatorsWithTwoServersAreRolled()
@@ -192,8 +270,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       locator2 = rollLocatorToCurrent(locator2, hostName, locatorPorts[1], getTestMethodName(),
           locatorString);
 
-      server1 = rollServerToCurrentAndCreateRegion(server1, regionType, null, shortcut.name(),
-          regionName, locatorPorts);
+      server1 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server1, regionType,
null,
+          shortcut.name(), regionName, locatorPorts);
       expectedRegionSize += 10;
       putSerializableObjectAndVerifyLuceneQueryResult(server2, regionName, expectedRegionSize,
15,
           25, server1, server2);
@@ -201,8 +279,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       putSerializableObjectAndVerifyLuceneQueryResult(server1, regionName, expectedRegionSize,
20,
           30, server1, server2);
 
-      server2 = rollServerToCurrentAndCreateRegion(server2, regionType, null, shortcut.name(),
-          regionName, locatorPorts);
+      server2 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server2, regionType,
null,
+          shortcut.name(), regionName, locatorPorts);
       expectedRegionSize += 5;
       putSerializableObjectAndVerifyLuceneQueryResult(server2, regionName, expectedRegionSize,
25,
           35, server1, server2);
@@ -281,8 +359,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       locator = rollLocatorToCurrent(locator, hostName, locatorPorts[0], getTestMethodName(),
           locatorString);
 
-      server3 = rollServerToCurrentAndCreateRegion(server3, regionType, null, shortcut.name(),
-          regionName, locatorPorts);
+      server3 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server3, regionType,
null,
+          shortcut.name(), regionName, locatorPorts);
       invokeRunnableInVMs(invokeStartCacheServer(csPorts[1]), server3);
       expectedRegionSize += 10;
       putSerializableObjectAndVerifyLuceneQueryResult(client, regionName, expectedRegionSize,
20,
@@ -291,8 +369,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       putSerializableObjectAndVerifyLuceneQueryResult(server3, regionName, expectedRegionSize,
30,
           40, server2);
 
-      server2 = rollServerToCurrentAndCreateRegion(server2, regionType, null, shortcut.name(),
-          regionName, locatorPorts);
+      server2 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server2, regionType,
null,
+          shortcut.name(), regionName, locatorPorts);
       invokeRunnableInVMs(invokeStartCacheServer(csPorts[0]), server2);
       expectedRegionSize += 10;
       putSerializableObjectAndVerifyLuceneQueryResult(client, regionName, expectedRegionSize,
40,
@@ -408,8 +486,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       // Roll the locator and server 1 to current version
       locator = rollLocatorToCurrent(locator, hostName, locatorPorts[0], getTestMethodName(),
           locatorString);
-      server1 = rollServerToCurrentAndCreateRegion(server1, regionType, null, shortcut.name(),
-          regionName, locatorPorts);
+      server1 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server1, regionType,
null,
+          shortcut.name(), regionName, locatorPorts);
 
       // Execute a query on the client and verify the results. This also waits until flushed.
       client.invoke(() -> verifyLuceneQueryResults(regionName, numObjects));
@@ -567,8 +645,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       locator = rollLocatorToCurrent(locator, hostName, locatorPorts[0], getTestMethodName(),
           locatorString);
 
-      server1 = rollServerToCurrentAndCreateRegion(server1, regionType, testingDirs[0],
-          shortcutName, regionName, locatorPorts);
+      server1 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server1, regionType,
+          testingDirs[0], shortcutName, regionName, locatorPorts);
       verifyLuceneQueryResultInEachVM(regionName, expectedRegionSize, server1);
       expectedRegionSize += 5;
       putSerializableObjectAndVerifyLuceneQueryResult(server1, regionName, expectedRegionSize,
5,
@@ -577,8 +655,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       putSerializableObjectAndVerifyLuceneQueryResult(server2, regionName, expectedRegionSize,
10,
           20, server1, server3);
 
-      server2 = rollServerToCurrentAndCreateRegion(server2, regionType, testingDirs[1],
-          shortcutName, regionName, locatorPorts);
+      server2 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server2, regionType,
+          testingDirs[1], shortcutName, regionName, locatorPorts);
       verifyLuceneQueryResultInEachVM(regionName, expectedRegionSize, server2);
       expectedRegionSize += 5;
       putSerializableObjectAndVerifyLuceneQueryResult(server2, regionName, expectedRegionSize,
15,
@@ -587,8 +665,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
       putSerializableObjectAndVerifyLuceneQueryResult(server3, regionName, expectedRegionSize,
20,
           30, server2, server3);
 
-      server3 = rollServerToCurrentAndCreateRegion(server3, regionType, testingDirs[2],
-          shortcutName, regionName, locatorPorts);
+      server3 = rollServerToCurrentCreateLuceneIndexAndCreateRegion(server3, regionType,
+          testingDirs[2], shortcutName, regionName, locatorPorts);
       verifyLuceneQueryResultInEachVM(regionName, expectedRegionSize, server3);
       putSerializableObjectAndVerifyLuceneQueryResult(server3, regionName, expectedRegionSize,
15,
           25, server1, server2);
@@ -710,8 +788,8 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
     return rollServer;
   }
 
-  private VM rollServerToCurrentAndCreateRegion(VM oldServer, String regionType, File diskdir,
-      String shortcutName, String regionName, int[] locatorPorts) throws Exception {
+  private VM rollServerToCurrentCreateLuceneIndexAndCreateRegion(VM oldServer, String regionType,
+      File diskdir, String shortcutName, String regionName, int[] locatorPorts) throws Exception
{
     VM rollServer = rollServerToCurrent(oldServer, locatorPorts);
     rollServer.invoke(() -> createLuceneIndex(cache, regionName, INDEX_NAME));
     // recreate region on "rolled" server
@@ -726,6 +804,21 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
     return rollServer;
   }
 
+  private VM rollServerToCurrentAndCreateRegionOnly(VM oldServer, String regionType, File
diskdir,
+      String shortcutName, String regionName, int[] locatorPorts) throws Exception {
+    VM rollServer = rollServerToCurrent(oldServer, locatorPorts);
+    // recreate region on "rolled" server
+    if ((regionType.equals("persistentPartitioned"))) {
+      CacheSerializableRunnable runnable =
+          invokeCreatePersistentPartitionedRegion(regionName, diskdir);
+      invokeRunnableInVMs(runnable, rollServer);
+    } else {
+      invokeRunnableInVMs(invokeCreateRegion(regionName, shortcutName), rollServer);
+    }
+    rollServer.invoke(invokeRebalance());
+    return rollServer;
+  }
+
   private VM rollLocatorToCurrent(VM oldLocator, final String serverHostName, final int port,
       final String testName, final String locatorString) throws Exception {
     // Roll the locator
@@ -992,6 +1085,22 @@ public class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCa
         .invoke(luceneIndexFactory, indexName, regionName);
   }
 
+  public static void createLuceneIndexOnExistingRegion(Object cache, String regionName,
+      String indexName) throws Exception {
+    Class luceneServiceProvider = Thread.currentThread().getContextClassLoader()
+        .loadClass("org.apache.geode.cache.lucene.LuceneServiceProvider");
+    Method getLuceneService = luceneServiceProvider.getMethod("get", GemFireCache.class);
+    Object luceneService = getLuceneService.invoke(luceneServiceProvider, cache);
+    Method createLuceneIndexFactoryMethod =
+        luceneService.getClass().getMethod("createIndexFactory");
+    createLuceneIndexFactoryMethod.setAccessible(true);
+    Object luceneIndexFactory = createLuceneIndexFactoryMethod.invoke(luceneService);
+    luceneIndexFactory.getClass().getMethod("addField", String.class).invoke(luceneIndexFactory,
+        "status");
+    luceneIndexFactory.getClass().getMethod("create", String.class, String.class, boolean.class)
+        .invoke(luceneIndexFactory, indexName, regionName, true);
+  }
+
   public static void createPersistentPartitonedRegion(Object cache, String regionName,
       File diskStore) throws Exception {
     Object store = cache.getClass().getMethod("findDiskStore", String.class).invoke(cache,
"store");
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
index 8b09ff5..876811f 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
@@ -163,5 +163,10 @@ public class LuceneServiceImplJUnitTest {
           (PartitionedRegion) index.getCache().getRegion(index.getRegionPath());
       verify(userRegion, never()).addAsyncEventQueueId(anyString(), anyBoolean());
     }
+
+    @Override
+    protected void validateAllMembersAreTheSameVersion(PartitionedRegion region) {
+
+    }
   }
 }

-- 
To stop receiving notification emails like this one, please contact
nnag@apache.org.

Mime
View raw message