You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by on...@apache.org on 2021/03/04 02:35:32 UTC
[geode] branch support/1.14 updated: GEODE-8996: Fixed rebalance
gfsh and REST API in mixed version mode (#6080)
This is an automated email from the ASF dual-hosted git repository.
onichols pushed a commit to branch support/1.14
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/support/1.14 by this push:
new b7a931e GEODE-8996: Fixed rebalance gfsh and REST API in mixed version mode (#6080)
b7a931e is described below
commit b7a931e4bc55694438da8e7ec61b79d206d96b5b
Author: Nabarun Nag <na...@users.noreply.github.com>
AuthorDate: Wed Mar 3 17:08:59 2021 -0800
GEODE-8996: Fixed rebalance gfsh and REST API in mixed version mode (#6080)
* Moved new child RebalanceFunction and CacheRealizationFunction to pre 1.12.0 locations.
* While talking to pre-1.12.0 servers, the locators send the function from the old package.
* While talking to 1.12.0 server, the new package function is used.
* For RebalanceFunction and CacheRealizationFunction the serialVersionUID is set to the one created by 1.11.0 for old package location and serialVersionUID created by 1.12.0 for the latter.
(cherry picked from commit 3faf283c038880755a7356fe570a4f92a46826cd)
---
geode-assembly/build.gradle | 3 +-
.../web/controllers/RestAPICompatibilityTest.java | 201 +++++++++++++++++++++
.../api/LocatorClusterManagementService.java | 48 ++++-
.../cli/functions/CacheRealizationFunction.java | 23 +++
.../internal/cli/functions/RebalanceFunction.java | 22 +++
.../functions/CacheRealizationFunction.java | 1 +
.../operation/RebalanceOperationPerformer.java | 34 +++-
.../sanctioned-geode-core-serializables.txt | 4 +-
.../operation/RebalanceOperationPerformerTest.java | 4 +-
.../internal/cli/commands/RebalanceCommand.java | 3 +-
.../geode/management/GfshCompatibilityTest.java | 29 ++-
11 files changed, 354 insertions(+), 18 deletions(-)
diff --git a/geode-assembly/build.gradle b/geode-assembly/build.gradle
index 51af352..8e11e7f 100755
--- a/geode-assembly/build.gradle
+++ b/geode-assembly/build.gradle
@@ -297,8 +297,9 @@ dependencies {
upgradeTestCompileOnly(platform(project(':boms:geode-all-bom')))
upgradeTestCompileOnly('io.swagger:swagger-annotations')
-
upgradeTestRuntimeOnly(project(path: ':geode-old-versions', configuration: 'testOutput'))
+ distributedTestRuntimeOnly(project(path: ':geode-old-versions', configuration: 'testOutput'))
+ testImplementation('org.assertj:assertj-core')
upgradeTestRuntimeOnly(project(':extensions:session-testing-war'))
upgradeTestRuntimeOnly('org.codehaus.cargo:cargo-core-uberjar')
upgradeTestRuntimeOnly('org.apache.httpcomponents:httpclient')
diff --git a/geode-assembly/src/distributedTest/java/org/apache/geode/rest/internal/web/controllers/RestAPICompatibilityTest.java b/geode-assembly/src/distributedTest/java/org/apache/geode/rest/internal/web/controllers/RestAPICompatibilityTest.java
new file mode 100644
index 0000000..11ba362
--- /dev/null
+++ b/geode-assembly/src/distributedTest/java/org/apache/geode/rest/internal/web/controllers/RestAPICompatibilityTest.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ *
+ */
+package org.apache.geode.rest.internal.web.controllers;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.management.configuration.DiskStore;
+import org.apache.geode.management.operation.RebalanceOperation;
+import org.apache.geode.management.operation.RestoreRedundancyRequest;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.BackwardCompatibilityTest;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.MemberStarterRule;
+import org.apache.geode.test.version.TestVersion;
+import org.apache.geode.test.version.VersionManager;
+import org.apache.geode.util.internal.GeodeJsonMapper;
+
+@Category({BackwardCompatibilityTest.class})
+@RunWith(Parameterized.class)
+public class RestAPICompatibilityTest {
+ private final String oldVersion;
+ private static ObjectMapper mapper = GeodeJsonMapper.getMapper();
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<String> data() {
+ List<String> result = VersionManager.getInstance().getVersionsWithoutCurrent();
+ result.removeIf(s -> TestVersion.compare(s, "1.11.0") < 0);
+ return result;
+ }
+
+ @Rule
+ public ClusterStartupRule cluster = new ClusterStartupRule();
+
+ @Rule
+ public GfshCommandRule gfsh = new GfshCommandRule();
+
+ public RestAPICompatibilityTest(String oldVersion) throws JsonProcessingException {
+ this.oldVersion = oldVersion;
+ DiskStore diskStore = new DiskStore();
+ diskStore.setName("diskStore");
+ postRESTAPICalls = new HashMap<>();
+ // {REST endpoint,{Body, Successful Status Message, Introduced in version}}
+ postRESTAPICalls.put("/management/v1/operations/rebalances",
+ new String[] {mapper.writeValueAsString(new RebalanceOperation()), "Operation started",
+ "1.11.0"});
+ postRESTAPICalls.put("/management/v1/operations/restoreRedundancy",
+ new String[] {mapper.writeValueAsString(new RestoreRedundancyRequest()),
+ "Operation started", "1.13.1"});
+ }
+
+ private static Map<String, String[]> postRESTAPICalls;
+
+
+ private static final String[][] getRESTAPICalls = {
+ // REST endpoint , status
+ {"/geode-mgmt/v1/management/commands?cmd=rebalance", "OK"}
+ };
+
+ @Test
+ public void restCommandExecutedOnLatestLocatorShouldBeBackwardsCompatible() throws Exception {
+ // Initialize all cluster members with old versions
+ MemberVM locator1 =
+ cluster.startLocatorVM(0, 0, oldVersion, MemberStarterRule::withHttpService);
+ int locatorPort1 = locator1.getPort();
+ MemberVM locator2 =
+ cluster.startLocatorVM(1, 0, oldVersion,
+ x -> x.withConnectionToLocator(locatorPort1).withHttpService());
+ int locatorPort2 = locator2.getPort();
+ cluster
+ .startServerVM(2, oldVersion, s -> s.withRegion(RegionShortcut.PARTITION, "region")
+ .withConnectionToLocator(locatorPort1, locatorPort2));
+ cluster
+ .startServerVM(3, oldVersion, s -> s.withRegion(RegionShortcut.PARTITION, "region")
+ .withConnectionToLocator(locatorPort1, locatorPort2));
+
+ // Roll locators to the current version
+ cluster.stop(0);
+ // gradle sets a property telling us where the build is located
+ final String buildDir = System.getProperty("geode.build.dir", System.getProperty("user.dir"));
+ locator1 = cluster.startLocatorVM(0, l -> l.withHttpService().withPort(locatorPort1)
+ .withConnectionToLocator(locatorPort2)
+ .withSystemProperty("geode.build.dir", buildDir));
+ cluster.stop(1);
+
+ cluster.startLocatorVM(1,
+ x -> x.withConnectionToLocator(locatorPort1).withHttpService().withPort(locatorPort2)
+ .withConnectionToLocator(locatorPort1)
+ .withSystemProperty("geode.build.dir", buildDir));
+
+ gfsh.connectAndVerify(locator1);
+ gfsh.execute("list members");
+ // Execute REST api calls to from the new locators to the old servers to ensure that backwards
+ // compatibility is maintained
+
+ executeAndValidatePOSTRESTCalls(locator1.getHttpPort());
+ executeAndValidateGETRESTCalls(locator1.getHttpPort());
+
+ }
+
+ void executeAndValidatePOSTRESTCalls(int locator) throws Exception {
+
+ try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
+ for (Map.Entry<String, String[]> entry : postRESTAPICalls.entrySet()) {
+ // Skip the test is the version is before the REST api was introduced.
+ if (TestVersion.compare(oldVersion, entry.getValue()[2]) < 0) {
+ continue;
+ }
+ HttpPost post =
+ new HttpPost("http://localhost:" + locator + entry.getKey());
+ post.addHeader("Content-Type", "application/json");
+ post.addHeader("Accept", "application/json");
+ StringEntity jsonStringEntity =
+ new StringEntity(entry.getValue()[0], ContentType.DEFAULT_TEXT);
+ post.setEntity(jsonStringEntity);
+ CloseableHttpResponse response = httpClient.execute(post);
+
+ HttpEntity entity = response.getEntity();
+ InputStream content = entity.getContent();
+
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(content))) {
+ String line;
+ StringBuilder sb = new StringBuilder();
+ while ((line = reader.readLine()) != null) {
+ sb.append(line);
+ }
+ JsonNode jsonObject = mapper.readTree(sb.toString());
+ String statusCode = jsonObject.findValue("statusCode").textValue();
+ assertThat(statusCode).satisfiesAnyOf(
+ value -> assertThat(value).isEqualTo("ACCEPTED"),
+ value -> assertThat(value).contains("OK"));
+ String statusMessage = jsonObject.findValue("statusMessage").textValue();
+ assertThat(statusMessage).contains(entry.getValue()[1]);
+ }
+ }
+ }
+ }
+
+ public static void executeAndValidateGETRESTCalls(int locator) throws Exception {
+
+ try (CloseableHttpClient httpclient = HttpClients.createDefault()) {
+ for (String[] commandExpectedResponsePair : getRESTAPICalls) {
+ HttpGet get =
+ new HttpGet("http://localhost:" + locator +
+ commandExpectedResponsePair[0]);
+ CloseableHttpResponse response = httpclient.execute(get);
+ HttpEntity entity = response.getEntity();
+ InputStream content = entity.getContent();
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(content))) {
+ String line;
+ StringBuilder sb = new StringBuilder();
+ while ((line = reader.readLine()) != null) {
+ sb.append(line);
+ }
+ JsonNode jsonObject = mapper.readTree(sb.toString());
+ String statusCode = jsonObject.findValue("status").textValue();
+ assertThat(statusCode).contains(commandExpectedResponsePair[1]);
+ }
+ }
+ }
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/api/LocatorClusterManagementService.java b/geode-core/src/main/java/org/apache/geode/management/internal/api/LocatorClusterManagementService.java
index f8d9f52..3c81dab 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/api/LocatorClusterManagementService.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/api/LocatorClusterManagementService.java
@@ -49,8 +49,10 @@ import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.InternalConfigurationPersistenceService;
import org.apache.geode.distributed.internal.locks.DLockService;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.execute.AbstractExecution;
+import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.api.ClusterManagementException;
@@ -595,8 +597,18 @@ public class LocatorClusterManagementService implements ClusterManagementService
if (targetMembers.size() == 0) {
return Collections.emptyList();
}
+ Set<DistributedMember> targetMemberPRE1_12_0 = new HashSet<>();
+ Set<DistributedMember> targetMemberPOST1_12_0 = new HashSet<>();
+
+ targetMembers.stream().forEach(member -> {
+ if (((InternalDistributedMember) member).getVersion()
+ .isOlderThan(KnownVersion.GEODE_1_12_0)) {
+ targetMemberPRE1_12_0.add(member);
+ } else {
+ targetMemberPOST1_12_0.add(member);
+ }
+ });
- Function function = new CacheRealizationFunction();
File file = null;
@@ -605,11 +617,24 @@ public class LocatorClusterManagementService implements ClusterManagementService
}
if (file == null) {
- Execution execution = FunctionService.onMembers(targetMembers)
- .setArguments(Arrays.asList(configuration, operation, null));
- ((AbstractExecution) execution).setIgnoreDepartedMembers(true);
- List<?> functionResults = (List<?>) execution.execute(function).getResult();
- return cleanResults(functionResults);
+ List<?> functionResults = new ArrayList<>();
+ if (targetMemberPRE1_12_0.size() > 0) {
+ Function function =
+ new org.apache.geode.management.internal.cli.functions.CacheRealizationFunction();
+ Execution execution = FunctionService.onMembers(targetMemberPRE1_12_0)
+ .setArguments(Arrays.asList(configuration, operation, null));
+ ((AbstractExecution) execution).setIgnoreDepartedMembers(true);
+ functionResults.addAll(cleanResults((List<?>) execution.execute(function).getResult()));
+ }
+ if (targetMemberPOST1_12_0.size() > 0) {
+ Function function = new CacheRealizationFunction();
+ Execution execution = FunctionService.onMembers(targetMemberPOST1_12_0)
+ .setArguments(Arrays.asList(configuration, operation, null));
+ ((AbstractExecution) execution).setIgnoreDepartedMembers(true);
+ functionResults.addAll(cleanResults((List<?>) execution.execute(function).getResult()));
+ }
+
+ return (List<R>) functionResults;
}
// if we have file arguments, we need to export the file input stream for each member
@@ -631,7 +656,16 @@ public class LocatorClusterManagementService implements ClusterManagementService
Execution execution = FunctionService.onMember(member)
.setArguments(Arrays.asList(configuration, operation, remoteInputStream));
((AbstractExecution) execution).setIgnoreDepartedMembers(true);
- List<R> functionResults = cleanResults((List<?>) execution.execute(function).getResult());
+ List<R> functionResults;
+ if (((InternalDistributedMember) member).getVersion()
+ .isOlderThan(KnownVersion.GEODE_1_12_0)) {
+ Function function =
+ new org.apache.geode.management.internal.cli.functions.CacheRealizationFunction();
+ functionResults = cleanResults((List<?>) execution.execute(function).getResult());
+ } else {
+ Function function = new CacheRealizationFunction();
+ functionResults = cleanResults((List<?>) execution.execute(function).getResult());
+ }
results.addAll(functionResults);
} catch (IOException e) {
raise(StatusCode.ILLEGAL_ARGUMENT, "Invalid file: " + file.getAbsolutePath());
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CacheRealizationFunction.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CacheRealizationFunction.java
new file mode 100644
index 0000000..9e8caf0
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CacheRealizationFunction.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.geode.management.internal.cli.functions;
+
+public class CacheRealizationFunction extends
+ org.apache.geode.management.internal.functions.CacheRealizationFunction {
+ private static final long serialVersionUID = 6209080805559452304L;
+}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/RebalanceFunction.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/RebalanceFunction.java
new file mode 100644
index 0000000..9d6bf39
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/RebalanceFunction.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.geode.management.internal.cli.functions;
+
+public class RebalanceFunction extends
+ org.apache.geode.management.internal.functions.RebalanceFunction {
+ private static final long serialVersionUID = 1L;
+}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/functions/CacheRealizationFunction.java b/geode-core/src/main/java/org/apache/geode/management/internal/functions/CacheRealizationFunction.java
index 3e8733a..0c330fa 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/functions/CacheRealizationFunction.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/functions/CacheRealizationFunction.java
@@ -68,6 +68,7 @@ public class CacheRealizationFunction implements InternalFunction<List> {
private static final Logger logger = LogService.getLogger();
@Immutable
private static final Map<Class, ConfigurationRealizer> realizers = new HashMap<>();
+ private static final long serialVersionUID = -2695517414081975343L;
static {
realizers.put(Region.class, new RegionConfigRealizer());
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformer.java b/geode-core/src/main/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformer.java
index 1ab09fd..21f2d00 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformer.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformer.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Set;
import org.apache.commons.collections.CollectionUtils;
+import org.jetbrains.annotations.NotNull;
import org.apache.geode.annotations.Experimental;
import org.apache.geode.annotations.VisibleForTesting;
@@ -37,7 +38,9 @@ import org.apache.geode.cache.control.RebalanceResults;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.management.DistributedRegionMXBean;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.internal.MBeanJMXAdapter;
@@ -123,11 +126,11 @@ public class RebalanceOperationPerformer
functionArgs[1] = setRegionName;
functionArgs[2] = null;
-
+ Function function = getRebalanceFunction((InternalDistributedMember) member);
List<String> resultList = null;
try {
resultList = (List<String>) ManagementUtils
- .executeFunction(new RebalanceFunction(), functionArgs, Collections.singleton(member))
+ .executeFunction(function, functionArgs, Collections.singleton(member))
.getResult();
} catch (Exception ignored) {
@@ -136,7 +139,6 @@ public class RebalanceOperationPerformer
RebalanceRegionResult result = new RebalanceRegionResultImpl();
if (resultList != null && !resultList.isEmpty()) {
List<String> rstList = Arrays.asList(resultList.get(0).split(","));
-
result = toRebalanceRegionResult(rstList);
}
@@ -375,7 +377,8 @@ public class RebalanceOperationPerformer
if (memberPR.dsMemberList.size() > 1) {
for (int i = 0; i < memberPR.dsMemberList.size(); i++) {
DistributedMember dsMember = memberPR.dsMemberList.get(i);
- RebalanceFunction rebalanceFunction = new RebalanceFunction();
+ Function rebalanceFunction = getRebalanceFunction(
+ (InternalDistributedMember) dsMember);
Object[] functionArgs = new Object[3];
functionArgs[0] = simulate;
Set<String> regionSet = new HashSet<>();
@@ -445,6 +448,19 @@ public class RebalanceOperationPerformer
return rebalanceResult;
}
+ @NotNull
+ private Function getRebalanceFunction(InternalDistributedMember dsMember) {
+ Function rebalanceFunction;
+ if (dsMember.getVersion()
+ .isOlderThan(KnownVersion.GEODE_1_12_0)) {
+ rebalanceFunction =
+ new org.apache.geode.management.internal.cli.functions.RebalanceFunction();
+ } else {
+ rebalanceFunction = new RebalanceFunction();
+ }
+ return rebalanceFunction;
+ }
+
private static RebalanceRegionResult toRebalanceRegionResult(List<String> rstList) {
RebalanceRegionResultImpl result = new RebalanceRegionResultImpl();
result.setBucketCreateBytes(Long.parseLong(rstList.get(0)));
@@ -456,8 +472,14 @@ public class RebalanceOperationPerformer
result.setPrimaryTransferTimeInMilliseconds(Long.parseLong(rstList.get(6)));
result.setPrimaryTransfersCompleted(Integer.parseInt(rstList.get(7)));
result.setTimeInMilliseconds(Long.parseLong(rstList.get(8)));
- result.setNumOfMembers(Integer.parseInt(rstList.get(9)));
- result.setRegionName(rstList.get(10).replace(SEPARATOR, ""));
+ if (rstList.size() < 11) {
+ result.setNumOfMembers(-1);
+ result.setRegionName(rstList.get(9).replace(SEPARATOR, ""));
+ } else {
+ result.setNumOfMembers(Integer.parseInt(rstList.get(9)));
+ result.setRegionName(rstList.get(10).replace(SEPARATOR, ""));
+ }
+
return result;
}
diff --git a/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt b/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt
index 652d1b2..356102c 100644
--- a/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt
+++ b/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt
@@ -443,6 +443,8 @@ org/apache/geode/management/internal/beans/FileUploader$RemoteFile,false,filenam
org/apache/geode/management/internal/beans/QueryDataFunction,true,1
org/apache/geode/management/internal/beans/QueryDataFunction$LocalQueryFunction,true,1,id:java/lang/String,optimizeForWrite:boolean,regionName:java/lang/String,showMembers:boolean,this$0:org/apache/geode/management/internal/beans/QueryDataFunction
org/apache/geode/management/internal/beans/stats/StatType,false
+org/apache/geode/management/internal/cli/functions/CacheRealizationFunction,true,6209080805559452304
+org/apache/geode/management/internal/cli/functions/RebalanceFunction,true,1
org/apache/geode/management/internal/configuration/domain/SharedConfigurationStatus,false
org/apache/geode/management/internal/configuration/functions/DownloadJarFunction,true,1
org/apache/geode/management/internal/configuration/functions/GetClusterConfigurationFunction,false
@@ -453,7 +455,7 @@ org/apache/geode/management/internal/configuration/messages/ClusterManagementSer
org/apache/geode/management/internal/exceptions/EntityExistsException,false
org/apache/geode/management/internal/exceptions/EntityNotFoundException,false,statusOK:boolean
org/apache/geode/management/internal/exceptions/NoMembersException,false
-org/apache/geode/management/internal/functions/CacheRealizationFunction,false
+org/apache/geode/management/internal/functions/CacheRealizationFunction,true,-2695517414081975343
org/apache/geode/management/internal/functions/CliFunctionResult$StatusState,false
org/apache/geode/management/internal/functions/GetMemberInformationFunction,true,1404642539058875565
org/apache/geode/management/internal/functions/RebalanceFunction,true,1
diff --git a/geode-core/src/test/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformerTest.java b/geode-core/src/test/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformerTest.java
index 0f20fe14..21ef1e4 100644
--- a/geode-core/src/test/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformerTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/internal/operation/RebalanceOperationPerformerTest.java
@@ -36,6 +36,7 @@ import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalCacheForClientAccess;
+import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.management.DistributedRegionMXBean;
import org.apache.geode.management.DistributedSystemMXBean;
import org.apache.geode.management.ManagementService;
@@ -137,7 +138,8 @@ public class RebalanceOperationPerformerTest {
List<Object> resultList = new ArrayList<>();
resultList.add("0,1,2,3,4,5,6,7,8,9," + SEPARATOR + "region1");
when(functionExecutor.execute(any(), any(), any())).thenReturn(resultList);
-
+ when(distributedMember1.getVersion()).thenReturn(KnownVersion.getCurrentVersion());
+ when(distributedMember2.getVersion()).thenReturn(KnownVersion.getCurrentVersion());
RebalanceResult result =
performer.executeRebalanceOnDS(managementService, cache, "true",
Collections.emptyList(), functionExecutor);
diff --git a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java
index 62cc54c..c1dd8c7 100644
--- a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java
+++ b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/RebalanceCommand.java
@@ -107,7 +107,8 @@ public class RebalanceCommand extends GfshCommand {
rsltList.add(6, String.valueOf(results.getPrimaryTransferTimeInMilliseconds()));
rsltList.add(7, String.valueOf(results.getPrimaryTransfersCompleted()));
rsltList.add(8, String.valueOf(results.getTimeInMilliseconds()));
- rsltList.add(9, String.valueOf(results.getNumOfMembers()));
+ rsltList.add(9, results.getNumOfMembers() == -1 ? "Not Available"
+ : String.valueOf(results.getNumOfMembers()));
String regionName = results.getRegionName();
if (!regionName.startsWith(SEPARATOR)) {
regionName = SEPARATOR + regionName;
diff --git a/geode-gfsh/src/upgradeTest/java/org/apache/geode/management/GfshCompatibilityTest.java b/geode-gfsh/src/upgradeTest/java/org/apache/geode/management/GfshCompatibilityTest.java
index 07690ec..494cdb9 100644
--- a/geode-gfsh/src/upgradeTest/java/org/apache/geode/management/GfshCompatibilityTest.java
+++ b/geode-gfsh/src/upgradeTest/java/org/apache/geode/management/GfshCompatibilityTest.java
@@ -26,6 +26,7 @@ import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.test.dunit.rules.ClusterStartupRule;
import org.apache.geode.test.dunit.rules.MemberVM;
import org.apache.geode.test.junit.categories.BackwardCompatibilityTest;
@@ -43,6 +44,7 @@ public class GfshCompatibilityTest {
@Parameterized.Parameters(name = "{0}")
public static Collection<String> data() {
List<String> result = VersionManager.getInstance().getVersionsWithoutCurrent();
+ result.removeIf(s -> TestVersion.compare(s, "1.11.0") < 0);
return result;
}
@@ -70,7 +72,7 @@ public class GfshCompatibilityTest {
assertThat(gfsh.isConnected()).isFalse();
assertThat(gfsh.getGfshOutput()).contains("Cannot use a")
.contains("gfsh client to connect to this cluster.");
- } else if (TestVersion.compare(oldVersion, "1.10.0") < 0) {
+ } else if (TestVersion.compare(oldVersion, "1.11.0") < 0) {
gfsh.connect(oldLocator.getPort(), GfshCommandRule.PortType.locator);
assertThat(gfsh.isConnected()).isFalse();
assertThat(gfsh.getGfshOutput()).contains("Cannot use a")
@@ -86,4 +88,29 @@ public class GfshCompatibilityTest {
}
}
+ @Test
+ public void whenCurrentVersionLocatorsExecuteRebalanceOnOldServersThenItMustSucceed()
+ throws Exception {
+ MemberVM locator1 = cluster.startLocatorVM(0, oldVersion);
+ int locatorPort1 = locator1.getPort();
+ MemberVM locator2 =
+ cluster.startLocatorVM(1, 0, oldVersion, x -> x.withConnectionToLocator(locatorPort1));
+ int locatorPort2 = locator2.getPort();
+ cluster
+ .startServerVM(2, oldVersion, s -> s.withRegion(RegionShortcut.PARTITION, "region")
+ .withConnectionToLocator(locatorPort1, locatorPort2));
+ cluster
+ .startServerVM(3, oldVersion, s -> s.withRegion(RegionShortcut.PARTITION, "region")
+ .withConnectionToLocator(locatorPort1, locatorPort2));
+ cluster.stop(0);
+ locator1 = cluster.startLocatorVM(0, x -> x.withConnectionToLocator(locatorPort2));
+ cluster.stop(1);
+ int locatorPort1_v2 = locator1.getPort();
+ cluster.startLocatorVM(1, x -> x.withConnectionToLocator(locatorPort1_v2));
+ gfsh.connectAndVerify(locator1);
+ gfsh.executeAndAssertThat("rebalance ")
+ .statusIsSuccess();
+
+ }
+
}