DonalEvans commented on a change in pull request #6601:
URL: https://github.com/apache/geode/pull/6601#discussion_r678478995
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
##########
@@ -426,8 +428,12 @@ public void cmdExecute(final Message clientMessage, final
ServerConnection serve
if (isPdxEvent) {
result = addPdxType(crHelper, key, value);
} else {
+ boolean generateCallbacks = true;
+ if (actionType ==
GatewaySenderEventImpl.UPDATE_ACTION_NO_GENERATE_CALLBACKS) {
+ generateCallbacks = false;
+ }
Review comment:
This can be simplified to
```
boolean generateCallbacks = actionType !=
GatewaySenderEventImpl.UPDATE_ACTION_NO_GENERATE_CALLBACKS;
```
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
##########
@@ -5180,7 +5180,8 @@ public boolean basicBridgeCreate(final Object key, final
byte[] value, boolean i
public boolean basicBridgePut(Object key, Object value, byte[] deltaBytes,
boolean isObject,
Object callbackArg, ClientProxyMembershipID memberId, boolean fromClient,
Review comment:
While you're modifying this method, the argument `fromClient` is never
used in the method. Could it be removed?
##########
File path:
geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunction.java
##########
@@ -0,0 +1,562 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.cache.Declarable;
+import org.apache.geode.cache.EntryDestroyedException;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.AllConnectionsInUseException;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import
org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DefaultEntryEventFactory;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EntrySnapshot;
+import org.apache.geode.internal.cache.EnumListenerEvent;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.NonTXEntry;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliFunction;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+/**
+ * Class for copying via WAN the contents of a region
+ * It must be executed in all members of the Geode cluster that host the region
+ * to be copied. (called with onServers() or withMembers() passing the list
+ * of all members hosting the region).
+ * It also offers the possibility to cancel an ongoing execution of this
function.
+ * The copying itself is executed in a new thread with a known name
+ * (parameterized with the regionName and senderId) in order to allow
+ * to cancel ongoing invocations by interrupting that thread.
+ *
+ * It accepts the following arguments in an array of objects
+ * 0: regionName (String)
+ * 1: senderId (String)
+ * 2: isCancel (Boolean): If true, it indicates that an ongoing execution of
this
+ * function for the given region and senderId must be stopped. Otherwise,
+ * it indicates that the region must be copied.
+ * 3: maxRate (Long) maximum copy rate in entries per second. In the case of
+ * parallel gateway senders, the maxRate is per server hosting the region.
+ * 4: batchSize (Integer): the size of the batches. Region entries are copied
in batches of the
+ * passed size. After each batch is sent, the function checks if the command
+ * must be canceled and also sleeps for some time if necessary to adjust the
+ * copy rate to the one passed as argument.
+ */
+public class WanCopyRegionFunction extends CliFunction<Object[]> implements
Declarable {
+ private static final Logger logger = LogService.getLogger();
+ private static final long serialVersionUID = 1L;
+
+ public static final String ID = WanCopyRegionFunction.class.getName();
+
+ private static final int MAX_BATCH_SEND_RETRIES = 1;
+
+ private static final int THREAD_POOL_SIZE = 10;
+
+ private final Clock clock;
+ private final ThreadSleeper threadSleeper;
+
+ private static final ExecutorService executor = LoggingExecutors
+ .newFixedThreadPool(THREAD_POOL_SIZE, "wanCopyRegionFunctionThread_",
true);
+
+ /**
+ * Contains the ongoing executions of this function
+ */
+ private static final Map<String, Future<?>> executions = new
ConcurrentHashMap<>();
+
+ private volatile int batchId = 0;
+
+ public WanCopyRegionFunction() {
+ this(Clock.systemDefaultZone(), new ThreadSleeper());
+ }
+
+ @VisibleForTesting
+ WanCopyRegionFunction(Clock clock, ThreadSleeper threadSleeper) {
+ this.clock = clock;
+ this.threadSleeper = threadSleeper;
+ }
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+ @Override
+ public boolean hasResult() {
+ return true;
+ }
+
+ @Override
+ public boolean isHA() {
+ return false;
+ }
+
+ @Override
+ public CliFunctionResult executeFunction(FunctionContext<Object[]> context) {
+ final Object[] args = context.getArguments();
+ if (args.length < 5) {
+ throw new IllegalStateException(
+ "Arguments length does not match required length.");
+ }
+ final String regionName = (String) args[0];
+ final String senderId = (String) args[1];
+ final boolean isCancel = (Boolean) args[2];
+ long maxRate = (Long) args[3];
+ int batchSize = (Integer) args[4];
+
+ if (regionName.endsWith("*") && senderId.equals("*") && isCancel) {
+ return cancelAllWanCopyRegion(context);
+ }
Review comment:
This functionality doesn't seem to be documented or tested. It would be
good to add Unit and DUnit tests for this, and document this special case.
##########
File path:
geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunction.java
##########
@@ -0,0 +1,562 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.cache.Declarable;
+import org.apache.geode.cache.EntryDestroyedException;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.AllConnectionsInUseException;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import
org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DefaultEntryEventFactory;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EntrySnapshot;
+import org.apache.geode.internal.cache.EnumListenerEvent;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.NonTXEntry;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliFunction;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+/**
+ * Class for copying via WAN the contents of a region
+ * It must be executed in all members of the Geode cluster that host the region
+ * to be copied. (called with onServers() or withMembers() passing the list
+ * of all members hosting the region).
+ * It also offers the possibility to cancel an ongoing execution of this
function.
+ * The copying itself is executed in a new thread with a known name
+ * (parameterized with the regionName and senderId) in order to allow
+ * to cancel ongoing invocations by interrupting that thread.
+ *
+ * It accepts the following arguments in an array of objects
+ * 0: regionName (String)
+ * 1: senderId (String)
+ * 2: isCancel (Boolean): If true, it indicates that an ongoing execution of
this
+ * function for the given region and senderId must be stopped. Otherwise,
+ * it indicates that the region must be copied.
+ * 3: maxRate (Long) maximum copy rate in entries per second. In the case of
+ * parallel gateway senders, the maxRate is per server hosting the region.
+ * 4: batchSize (Integer): the size of the batches. Region entries are copied
in batches of the
+ * passed size. After each batch is sent, the function checks if the command
+ * must be canceled and also sleeps for some time if necessary to adjust the
+ * copy rate to the one passed as argument.
+ */
+public class WanCopyRegionFunction extends CliFunction<Object[]> implements
Declarable {
+ private static final Logger logger = LogService.getLogger();
+ private static final long serialVersionUID = 1L;
+
+ public static final String ID = WanCopyRegionFunction.class.getName();
+
+ private static final int MAX_BATCH_SEND_RETRIES = 1;
+
+ private static final int THREAD_POOL_SIZE = 10;
+
+ private final Clock clock;
+ private final ThreadSleeper threadSleeper;
+
+ private static final ExecutorService executor = LoggingExecutors
+ .newFixedThreadPool(THREAD_POOL_SIZE, "wanCopyRegionFunctionThread_",
true);
+
+ /**
+ * Contains the ongoing executions of this function
+ */
+ private static final Map<String, Future<?>> executions = new
ConcurrentHashMap<>();
+
+ private volatile int batchId = 0;
+
+ public WanCopyRegionFunction() {
+ this(Clock.systemDefaultZone(), new ThreadSleeper());
+ }
+
+ @VisibleForTesting
+ WanCopyRegionFunction(Clock clock, ThreadSleeper threadSleeper) {
+ this.clock = clock;
+ this.threadSleeper = threadSleeper;
+ }
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+ @Override
+ public boolean hasResult() {
+ return true;
+ }
+
+ @Override
+ public boolean isHA() {
+ return false;
+ }
+
+ @Override
+ public CliFunctionResult executeFunction(FunctionContext<Object[]> context) {
+ final Object[] args = context.getArguments();
+ if (args.length < 5) {
+ throw new IllegalStateException(
+ "Arguments length does not match required length.");
+ }
+ final String regionName = (String) args[0];
+ final String senderId = (String) args[1];
+ final boolean isCancel = (Boolean) args[2];
+ long maxRate = (Long) args[3];
+ int batchSize = (Integer) args[4];
+
+ if (regionName.endsWith("*") && senderId.equals("*") && isCancel) {
+ return cancelAllWanCopyRegion(context);
+ }
+
+ if (isCancel) {
+ return cancelWanCopyRegion(context, regionName, senderId);
+ }
+
+ final InternalCache cache = (InternalCache) context.getCache();
Review comment:
This can be `final Cache cache`, which removes the need for a cast.
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventCallbackDispatcher.java
##########
@@ -184,4 +187,11 @@ public void stop() {
public void shutDownAckReaderConnection() {
// no op
}
+
+ @Override
+ public void sendBatch(List<GatewayQueueEvent<?, ?>> events, Connection
connection,
+ ExecutablePool senderPool, int batchId, boolean
removeFromQueueOnException)
+ throws BatchException70 {
Review comment:
`BatchException70` is never thrown from this method, so this can be
removed.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -2180,6 +2216,31 @@ public static int createServer(int locPort, int
maximumTimeBetweenPings) {
return port;
}
+ public static void createClientWithLocator(int port0, String host, String
regionName,
Review comment:
This method might be better named "createClientWithLocatorAndRegion".
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -2180,6 +2216,31 @@ public static int createServer(int locPort, int
maximumTimeBetweenPings) {
return port;
}
+ public static void createClientWithLocator(int port0, String host, String
regionName,
+ ClientRegionShortcut regionType) {
+ ClientCache cache = new ClientCacheFactory().addPoolLocator(host,
port0).create();
+
+ Region region =
Review comment:
Compiler warning here (and elsewhere in the class) can be fixed by using
`Region<Object, Object>`.
##########
File path:
geode-gfsh/src/integrationTest/java/org/apache/geode/management/internal/cli/GfshParserAutoCompletionIntegrationTest.java
##########
@@ -596,4 +596,20 @@ public void
testCompletionOffersTheFirstMandatoryOptionInAlphabeticalOrderForRem
assertThat(candidate.getFirstCandidate()).isEqualTo(buffer + "region");
}
+ @Test
+ public void
testCompletionOffersMandatoryOptionsInAlphabeticalOrderForWanCopyRegionWithSpace()
{
+ String buffer = "wan-copy region ";
+ CommandCandidate candidate = gfshParserRule.complete(buffer);
+ assertThat(candidate.getCandidates()).hasSize(2);
+ assertThat(candidate.getFirstCandidate()).isEqualTo(buffer + "--region");
Review comment:
This test should be expanded to also verify that the mandatory
`--sender-id` option auto-completes as expected, not just the first mandatory
`--region` option.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -1320,6 +1339,23 @@ public static void checkGatewayReceiverStats(int
processBatches, int eventsRecei
assertEquals(creates, gatewayReceiverStats.getCreateRequest());
}
+ public static List<Integer> getReceiverStats() {
Review comment:
Rather than returning a `List<Integer>` it might make things clearer to
return the `GatewayReceiverStats` from this method so that callers can call the
required methods on that. This helps avoid potentially unclear code like `int
receivedBatches = serverInB.invoke(() ->
WANTestBase.getReceiverStats().get(2));` since it's not at all obvious from
that code why 2 is the correct index to use.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -2180,6 +2216,31 @@ public static int createServer(int locPort, int
maximumTimeBetweenPings) {
return port;
}
+ public static void createClientWithLocator(int port0, String host, String
regionName,
+ ClientRegionShortcut regionType) {
+ ClientCache cache = new ClientCacheFactory().addPoolLocator(host,
port0).create();
+
+ Region region =
+ cache.createClientRegionFactory(regionType)
+ .create(regionName);
+
+ assertNotNull(region);
+ }
+
+ public static void createClientWithLocatorAndRegions(int port0, String host,
+ ClientRegionShortcut regionType, List<String> regions) {
+ ClientCache cache = new ClientCacheFactory().addPoolLocator(host,
port0).create();
+
+ for (String regionName : regions) {
+ Region region =
+ cache.createClientRegionFactory(regionType)
+ .create(regionName);
+
+ assertNotNull(region);
+ }
+ }
Review comment:
This method is never used and can be removed.
##########
File path:
geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunction.java
##########
@@ -0,0 +1,562 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.cache.Declarable;
+import org.apache.geode.cache.EntryDestroyedException;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.AllConnectionsInUseException;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import
org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DefaultEntryEventFactory;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EntrySnapshot;
+import org.apache.geode.internal.cache.EnumListenerEvent;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.NonTXEntry;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliFunction;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+/**
+ * Class for copying via WAN the contents of a region
+ * It must be executed in all members of the Geode cluster that host the region
+ * to be copied. (called with onServers() or withMembers() passing the list
+ * of all members hosting the region).
+ * It also offers the possibility to cancel an ongoing execution of this
function.
+ * The copying itself is executed in a new thread with a known name
+ * (parameterized with the regionName and senderId) in order to allow
+ * to cancel ongoing invocations by interrupting that thread.
+ *
+ * It accepts the following arguments in an array of objects
+ * 0: regionName (String)
+ * 1: senderId (String)
+ * 2: isCancel (Boolean): If true, it indicates that an ongoing execution of
this
+ * function for the given region and senderId must be stopped. Otherwise,
+ * it indicates that the region must be copied.
+ * 3: maxRate (Long) maximum copy rate in entries per second. In the case of
+ * parallel gateway senders, the maxRate is per server hosting the region.
+ * 4: batchSize (Integer): the size of the batches. Region entries are copied
in batches of the
+ * passed size. After each batch is sent, the function checks if the command
+ * must be canceled and also sleeps for some time if necessary to adjust the
+ * copy rate to the one passed as argument.
+ */
+public class WanCopyRegionFunction extends CliFunction<Object[]> implements
Declarable {
+ private static final Logger logger = LogService.getLogger();
+ private static final long serialVersionUID = 1L;
+
+ public static final String ID = WanCopyRegionFunction.class.getName();
+
+ private static final int MAX_BATCH_SEND_RETRIES = 1;
+
+ private static final int THREAD_POOL_SIZE = 10;
+
+ private final Clock clock;
+ private final ThreadSleeper threadSleeper;
+
+ private static final ExecutorService executor = LoggingExecutors
+ .newFixedThreadPool(THREAD_POOL_SIZE, "wanCopyRegionFunctionThread_",
true);
+
+ /**
+ * Contains the ongoing executions of this function
+ */
+ private static final Map<String, Future<?>> executions = new
ConcurrentHashMap<>();
+
+ private volatile int batchId = 0;
+
+ public WanCopyRegionFunction() {
+ this(Clock.systemDefaultZone(), new ThreadSleeper());
+ }
+
+ @VisibleForTesting
+ WanCopyRegionFunction(Clock clock, ThreadSleeper threadSleeper) {
+ this.clock = clock;
+ this.threadSleeper = threadSleeper;
+ }
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+ @Override
+ public boolean hasResult() {
+ return true;
+ }
+
+ @Override
+ public boolean isHA() {
+ return false;
+ }
+
+ @Override
+ public CliFunctionResult executeFunction(FunctionContext<Object[]> context) {
+ final Object[] args = context.getArguments();
+ if (args.length < 5) {
+ throw new IllegalStateException(
+ "Arguments length does not match required length.");
+ }
+ final String regionName = (String) args[0];
+ final String senderId = (String) args[1];
+ final boolean isCancel = (Boolean) args[2];
+ long maxRate = (Long) args[3];
+ int batchSize = (Integer) args[4];
+
+ if (regionName.endsWith("*") && senderId.equals("*") && isCancel) {
+ return cancelAllWanCopyRegion(context);
+ }
+
+ if (isCancel) {
+ return cancelWanCopyRegion(context, regionName, senderId);
+ }
+
+ final InternalCache cache = (InternalCache) context.getCache();
+
+ final Region<?, ?> region = cache.getRegion(regionName);
+ if (region == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__FOUND,
regionName));
+ }
+
+ GatewaySender sender = cache.getGatewaySender(senderId);
+ if (sender == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND,
senderId));
+ }
+
+ if
(!region.getAttributes().getGatewaySenderIds().contains(sender.getId())) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__USING_SENDER,
regionName,
+ senderId));
+ }
+
+ if (!sender.isRunning()) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__RUNNING,
senderId));
+ }
+
+ if (!sender.isParallel() && !((InternalGatewaySender) sender).isPrimary())
{
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.OK,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY,
+ senderId));
+ }
+
+ try {
+ return executeWanCopyRegionFunctionInNewThread(context, region,
regionName, sender, maxRate,
+ batchSize);
+ } catch (InterruptedException | CancellationException e) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+ CliStrings.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED);
+ } catch (ExecutionException e) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__EXECUTION__FAILED,
e.getMessage()));
+ }
+ }
+
+ private CliFunctionResult executeWanCopyRegionFunctionInNewThread(
+ FunctionContext<Object[]> context,
+ Region<?, ?> region, String regionName, GatewaySender sender, long
maxRate, int batchSize)
+ throws InterruptedException, ExecutionException, CancellationException {
+ String executionName = getExecutionName(regionName, sender.getId());
+ Callable<CliFunctionResult> callable =
+ new WanCopyRegionCallable(this, context, region, sender, maxRate,
batchSize);
+ FutureTask<CliFunctionResult> future = new FutureTask<>(callable);
+
+ if (executions.putIfAbsent(executionName, future) != null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__ALREADY__RUNNING__COMMAND,
+ regionName, sender.getId()));
+ }
Review comment:
There is no test coverage for this case. Could Unit and possibly DUnit
tests be added to confirm that if multiple identical `wan-copy region` commands
are executed at the same time, this error message is returned?
##########
File path:
geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunction.java
##########
@@ -0,0 +1,562 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.cache.Declarable;
+import org.apache.geode.cache.EntryDestroyedException;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.AllConnectionsInUseException;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import
org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DefaultEntryEventFactory;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EntrySnapshot;
+import org.apache.geode.internal.cache.EnumListenerEvent;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.NonTXEntry;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliFunction;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+/**
+ * Class for copying via WAN the contents of a region
+ * It must be executed in all members of the Geode cluster that host the region
+ * to be copied. (called with onServers() or withMembers() passing the list
+ * of all members hosting the region).
+ * It also offers the possibility to cancel an ongoing execution of this
function.
+ * The copying itself is executed in a new thread with a known name
+ * (parameterized with the regionName and senderId) in order to allow
+ * to cancel ongoing invocations by interrupting that thread.
+ *
+ * It accepts the following arguments in an array of objects
+ * 0: regionName (String)
+ * 1: senderId (String)
+ * 2: isCancel (Boolean): If true, it indicates that an ongoing execution of
this
+ * function for the given region and senderId must be stopped. Otherwise,
+ * it indicates that the region must be copied.
+ * 3: maxRate (Long) maximum copy rate in entries per second. In the case of
+ * parallel gateway senders, the maxRate is per server hosting the region.
+ * 4: batchSize (Integer): the size of the batches. Region entries are copied
in batches of the
+ * passed size. After each batch is sent, the function checks if the command
+ * must be canceled and also sleeps for some time if necessary to adjust the
+ * copy rate to the one passed as argument.
+ */
+public class WanCopyRegionFunction extends CliFunction<Object[]> implements
Declarable {
+ private static final Logger logger = LogService.getLogger();
+ private static final long serialVersionUID = 1L;
+
+ public static final String ID = WanCopyRegionFunction.class.getName();
+
+ private static final int MAX_BATCH_SEND_RETRIES = 1;
+
+ private static final int THREAD_POOL_SIZE = 10;
+
+ private final Clock clock;
+ private final ThreadSleeper threadSleeper;
+
+ private static final ExecutorService executor = LoggingExecutors
+ .newFixedThreadPool(THREAD_POOL_SIZE, "wanCopyRegionFunctionThread_",
true);
+
+ /**
+ * Contains the ongoing executions of this function
+ */
+ private static final Map<String, Future<?>> executions = new
ConcurrentHashMap<>();
Review comment:
This can be:
```
private static final Map<String, Future<CliFunctionResult>> executions = new
ConcurrentHashMap<>();
```
since we always expect the `Future` to return a `CliFunctionResult`. With
this change, you can also change uses of `Future<?>` on line 378 and line 391
to `Future<CliFunctionResult>`
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -2927,6 +2996,40 @@ public static void validateRegionSize(String regionName,
final int regionSize) {
}
}
+ public ArrayList getKeys(String regionName) {
+ final Region r = cache.getRegion(SEPARATOR + regionName);
+ assertNotNull(r);
+ return new ArrayList(r.keySet());
+
+ }
Review comment:
Compiler warnings here can be fixed by using:
```
public List<Object> getKeys(String regionName) {
final Region<Object, Object> r = cache.getRegion(SEPARATOR + regionName);
assertNotNull(r);
return new ArrayList<>(r.keySet());
}
```
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/WanCopyRegionCommandDUnitTest.java
##########
@@ -0,0 +1,1292 @@
+/*
+ * 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.commands;
+
+import static org.apache.geode.cache.Region.SEPARATOR;
+import static
org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
+import static
org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__BATCHSIZE;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__CANCEL;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MAXRATE;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__COPIED__ENTRIES;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__EXECUTION__CANCELED;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__FOUND;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__REGION;
+import static
org.apache.geode.management.internal.i18n.CliStrings.WAN_COPY_REGION__SENDERID;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import org.assertj.core.api.Condition;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.DataPolicy;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.Scope;
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.client.ClientRegionShortcut;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.cache.wan.WANTestBase;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.management.internal.cli.result.model.ResultModel;
+import org.apache.geode.management.internal.cli.util.CommandStringBuilder;
+import org.apache.geode.management.internal.i18n.CliStrings;
+import org.apache.geode.test.dunit.AsyncInvocation;
+import org.apache.geode.test.dunit.IgnoredException;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.junit.assertions.CommandResultAssert;
+import org.apache.geode.test.junit.categories.WanTest;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.LocatorLauncherStartupRule;
+
+@Category({WanTest.class})
+public class WanCopyRegionCommandDUnitTest extends WANTestBase {
+
+ private static final long serialVersionUID = 1L;
+
+ private enum Gateway {
+ SENDER, RECEIVER
+ }
+
+ public WanCopyRegionCommandDUnitTest() {
+ super();
+ }
+
+ @Test
+ public void testUnsuccessfulExecution_RegionNotFound() throws Exception {
+ List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+ VM serverInB = vm3;
+ VM serverInC = vm4;
+ VM client = vm8;
+ String senderIdInA = "B";
+ String senderIdInB = "C";
+
+ Integer senderLocatorPort = create3WanSitesAndClient(true, vm0,
+ vm1, vm2, serversInA, serverInB, serverInC, client,
+ senderIdInA, senderIdInB);
+
+ int wanCopyRegionBatchSize = 20;
+ String regionName = "foo";
+
+ // Execute wan-copy region command
+ GfshCommandRule gfsh = new GfshCommandRule();
+ gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+ String commandString = new CommandStringBuilder(WAN_COPY_REGION)
+ .addOption(WAN_COPY_REGION__REGION, regionName)
+ .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
+ .addOption(WAN_COPY_REGION__BATCHSIZE,
String.valueOf(wanCopyRegionBatchSize))
+ .getCommandString();
+
+ // Check command status and output
+ CommandResultAssert command =
+ verifyStatusIsError(gfsh.executeAndAssertThat(commandString));
+ String message =
+ CliStrings.format(WAN_COPY_REGION__MSG__REGION__NOT__FOUND,
+ Region.SEPARATOR + regionName);
+
command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+ .containsExactly(message, message, message);
+ }
+
+ @Test
+ public void testUnsuccessfulExecution_SenderNotFound() throws Exception {
+ List<VM> serversInA = Arrays.asList(vm5, vm6, vm7);
+ VM serverInB = vm3;
+ VM serverInC = vm4;
+ VM client = vm8;
+ String senderIdInA = "B";
+ String senderIdInB = "C";
+
+ Integer senderLocatorPort = create3WanSitesAndClient(true, vm0,
+ vm1, vm2, serversInA, serverInB, serverInC, client,
+ senderIdInA, senderIdInB);
+
+ int wanCopyRegionBatchSize = 20;
+ String regionName = getRegionName(true);
+
+ // Execute wan-copy region command
+ GfshCommandRule gfsh = new GfshCommandRule();
+ gfsh.connectAndVerify(senderLocatorPort, GfshCommandRule.PortType.locator);
+ String commandString = new CommandStringBuilder(WAN_COPY_REGION)
+ .addOption(WAN_COPY_REGION__REGION, regionName)
+ .addOption(WAN_COPY_REGION__SENDERID, senderIdInA)
+ .addOption(WAN_COPY_REGION__BATCHSIZE,
String.valueOf(wanCopyRegionBatchSize))
+ .getCommandString();
+
+ // Check command status and output
+ CommandResultAssert command =
+ verifyStatusIsError(gfsh.executeAndAssertThat(commandString));
+ String message =
+ CliStrings.format(WAN_COPY_REGION__MSG__SENDER__NOT__FOUND,
senderIdInA);
+
command.hasTableSection(ResultModel.MEMBER_STATUS_SECTION).hasColumn("Message")
+ .containsExactly(message, message, message);
+ }
+
+ @Test
+ public void
testUnsuccessfulExecutionWithPartitionedRegionAndParallelSender_ExceptionAtReceiver()
+ throws Exception {
+ testUnsuccessfulExecution_ExceptionAtReceiver(true, true);
+ }
+
+ public void testUnsuccessfulExecution_ExceptionAtReceiver(
Review comment:
This method is only used in one test, so it doesn't seem to need to be
extracted here.
##########
File path:
geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunction.java
##########
@@ -0,0 +1,562 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.cache.Declarable;
+import org.apache.geode.cache.EntryDestroyedException;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.AllConnectionsInUseException;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import
org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DefaultEntryEventFactory;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EntrySnapshot;
+import org.apache.geode.internal.cache.EnumListenerEvent;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.NonTXEntry;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliFunction;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+/**
+ * Class for copying via WAN the contents of a region
+ * It must be executed in all members of the Geode cluster that host the region
+ * to be copied. (called with onServers() or withMembers() passing the list
+ * of all members hosting the region).
+ * It also offers the possibility to cancel an ongoing execution of this
function.
+ * The copying itself is executed in a new thread with a known name
+ * (parameterized with the regionName and senderId) in order to allow
+ * to cancel ongoing invocations by interrupting that thread.
+ *
+ * It accepts the following arguments in an array of objects
+ * 0: regionName (String)
+ * 1: senderId (String)
+ * 2: isCancel (Boolean): If true, it indicates that an ongoing execution of
this
+ * function for the given region and senderId must be stopped. Otherwise,
+ * it indicates that the region must be copied.
+ * 3: maxRate (Long) maximum copy rate in entries per second. In the case of
+ * parallel gateway senders, the maxRate is per server hosting the region.
+ * 4: batchSize (Integer): the size of the batches. Region entries are copied
in batches of the
+ * passed size. After each batch is sent, the function checks if the command
+ * must be canceled and also sleeps for some time if necessary to adjust the
+ * copy rate to the one passed as argument.
+ */
+public class WanCopyRegionFunction extends CliFunction<Object[]> implements
Declarable {
+ private static final Logger logger = LogService.getLogger();
+ private static final long serialVersionUID = 1L;
+
+ public static final String ID = WanCopyRegionFunction.class.getName();
+
+ private static final int MAX_BATCH_SEND_RETRIES = 1;
+
+ private static final int THREAD_POOL_SIZE = 10;
+
+ private final Clock clock;
+ private final ThreadSleeper threadSleeper;
+
+ private static final ExecutorService executor = LoggingExecutors
+ .newFixedThreadPool(THREAD_POOL_SIZE, "wanCopyRegionFunctionThread_",
true);
+
+ /**
+ * Contains the ongoing executions of this function
+ */
+ private static final Map<String, Future<?>> executions = new
ConcurrentHashMap<>();
+
+ private volatile int batchId = 0;
+
+ public WanCopyRegionFunction() {
+ this(Clock.systemDefaultZone(), new ThreadSleeper());
+ }
+
+ @VisibleForTesting
+ WanCopyRegionFunction(Clock clock, ThreadSleeper threadSleeper) {
+ this.clock = clock;
+ this.threadSleeper = threadSleeper;
+ }
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+ @Override
+ public boolean hasResult() {
+ return true;
+ }
+
+ @Override
+ public boolean isHA() {
+ return false;
+ }
+
+ @Override
+ public CliFunctionResult executeFunction(FunctionContext<Object[]> context) {
+ final Object[] args = context.getArguments();
+ if (args.length < 5) {
+ throw new IllegalStateException(
+ "Arguments length does not match required length.");
+ }
+ final String regionName = (String) args[0];
+ final String senderId = (String) args[1];
+ final boolean isCancel = (Boolean) args[2];
+ long maxRate = (Long) args[3];
+ int batchSize = (Integer) args[4];
+
+ if (regionName.endsWith("*") && senderId.equals("*") && isCancel) {
+ return cancelAllWanCopyRegion(context);
+ }
+
+ if (isCancel) {
+ return cancelWanCopyRegion(context, regionName, senderId);
+ }
+
+ final InternalCache cache = (InternalCache) context.getCache();
+
+ final Region<?, ?> region = cache.getRegion(regionName);
+ if (region == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__FOUND,
regionName));
+ }
+
+ GatewaySender sender = cache.getGatewaySender(senderId);
+ if (sender == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND,
senderId));
+ }
+
+ if
(!region.getAttributes().getGatewaySenderIds().contains(sender.getId())) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__USING_SENDER,
regionName,
+ senderId));
+ }
+
+ if (!sender.isRunning()) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__RUNNING,
senderId));
+ }
+
+ if (!sender.isParallel() && !((InternalGatewaySender) sender).isPrimary())
{
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.OK,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY,
+ senderId));
+ }
+
+ try {
+ return executeWanCopyRegionFunctionInNewThread(context, region,
regionName, sender, maxRate,
+ batchSize);
+ } catch (InterruptedException | CancellationException e) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+ CliStrings.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED);
+ } catch (ExecutionException e) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__EXECUTION__FAILED,
e.getMessage()));
+ }
+ }
+
+ private CliFunctionResult executeWanCopyRegionFunctionInNewThread(
+ FunctionContext<Object[]> context,
+ Region<?, ?> region, String regionName, GatewaySender sender, long
maxRate, int batchSize)
+ throws InterruptedException, ExecutionException, CancellationException {
+ String executionName = getExecutionName(regionName, sender.getId());
+ Callable<CliFunctionResult> callable =
+ new WanCopyRegionCallable(this, context, region, sender, maxRate,
batchSize);
+ FutureTask<CliFunctionResult> future = new FutureTask<>(callable);
+
+ if (executions.putIfAbsent(executionName, future) != null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__ALREADY__RUNNING__COMMAND,
+ regionName, sender.getId()));
+ }
+
+ try {
+ executor.execute(future);
+ return future.get();
+ } finally {
+ executions.remove(executionName);
+ }
+ }
+
+ static class WanCopyRegionCallable implements Callable<CliFunctionResult> {
+ private final FunctionContext<Object[]> context;
+ private final Region<?, ?> region;
+ private final GatewaySender sender;
+ private final long maxRate;
+ private final int batchSize;
+ private final WanCopyRegionFunction function;
+
+ public WanCopyRegionCallable(final WanCopyRegionFunction function,
+ final FunctionContext<Object[]> context, final Region<?, ?> region,
+ final GatewaySender sender, final long maxRate,
+ final int batchSize) {
+ this.function = function;
+ this.context = context;
+ this.region = region;
+ this.sender = sender;
+ this.maxRate = maxRate;
+ this.batchSize = batchSize;
+ }
+
+ @Override
+ public CliFunctionResult call() throws Exception {
+ return function.wanCopyRegion(context, region, sender, maxRate,
batchSize);
+ }
+ }
+
+ @VisibleForTesting
+ CliFunctionResult wanCopyRegion(FunctionContext<Object[]> context, Region<?,
?> region,
+ GatewaySender sender, long maxRate, int batchSize) throws
InterruptedException {
+ ConnectionState connectionState = new ConnectionState();
+ int copiedEntries = 0;
+ final InternalCache cache = (InternalCache) context.getCache();
+ Iterator<?> entriesIter = getEntries(region, sender).iterator();
+ final long startTime = clock.millis();
+
+ try {
+ while (entriesIter.hasNext()) {
+ List<GatewayQueueEvent<?, ?>> batch =
+ createBatch((InternalRegion) region, sender, batchSize, cache,
entriesIter);
+ if (batch.size() == 0) {
+ continue;
+ }
+ Optional<CliFunctionResult> connectionError =
+ connectionState.connectIfNeeded(context, sender);
+ if (connectionError.isPresent()) {
+ return connectionError.get();
+ }
+ Optional<CliFunctionResult> error =
+ sendBatch(context, sender, batch, connectionState, copiedEntries);
+ if (error.isPresent()) {
+ return error.get();
+ }
+ copiedEntries += batch.size();
+ doPostSendBatchActions(startTime, copiedEntries, maxRate);
+ }
+ } finally {
+ connectionState.close();
+ }
+
+ if (region.isDestroyed()) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+ CliStrings.format(
+ CliStrings.WAN_COPY_REGION__MSG__ERROR__AFTER__HAVING__COPIED,
+ "Region destroyed",
+ copiedEntries));
+ }
+
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.OK,
+ CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__COPIED__ENTRIES,
+ copiedEntries));
+ }
+
+ private Optional<CliFunctionResult> sendBatch(FunctionContext<Object[]>
context,
+ GatewaySender sender, List<GatewayQueueEvent<?, ?>> batch,
+ ConnectionState connectionState, int copiedEntries) {
+ GatewaySenderEventDispatcher dispatcher =
+ ((AbstractGatewaySender) sender).getEventProcessor().getDispatcher();
+ int retries = 0;
+
+ while (true) {
+ try {
+ dispatcher.sendBatch(batch, connectionState.getConnection(),
+ connectionState.getSenderPool(), getAndIncrementBatchId(), true);
+ return Optional.empty();
+ } catch (BatchException70 e) {
+ return Optional.of(new CliFunctionResult(context.getMemberName(),
+ CliFunctionResult.StatusState.ERROR,
+ CliStrings.format(
+ CliStrings.WAN_COPY_REGION__MSG__ERROR__AFTER__HAVING__COPIED,
+ e.getExceptions().get(0).getCause(), copiedEntries)));
+ } catch (ConnectionDestroyedException | ServerConnectivityException e) {
+ Optional<CliFunctionResult> error =
+ connectionState.reconnect(context, retries++, copiedEntries, e);
+ if (error.isPresent()) {
+ return error;
+ }
+ }
+ }
+ }
+
+ List<GatewayQueueEvent<?, ?>> createBatch(InternalRegion region,
GatewaySender sender,
+ int batchSize, InternalCache cache, Iterator<?> iter) {
+ int batchIndex = 0;
+ List<GatewayQueueEvent<?, ?>> batch = new ArrayList<>();
+
+ while (iter.hasNext() && batchIndex < batchSize) {
+ GatewayQueueEvent<?, ?> event =
+ createGatewaySenderEvent(cache, region, sender, (Region.Entry<?, ?>)
iter.next());
+ if (event != null) {
+ batch.add(event);
+ batchIndex++;
+ }
+ }
+ return batch;
+ }
+
+ Set<?> getEntries(Region<?, ?> region, GatewaySender sender) {
+ if (region instanceof PartitionedRegion && sender.isParallel()) {
+ return ((PartitionedRegion)
region).getDataStore().getAllLocalBucketRegions()
+ .stream()
+ .flatMap(br -> ((Set<?>)
br.entrySet()).stream()).collect(Collectors.toSet());
+ }
+ return region.entrySet();
+ }
+
+ @VisibleForTesting
+ GatewayQueueEvent<?, ?> createGatewaySenderEvent(InternalCache cache,
+ InternalRegion region, GatewaySender sender, Region.Entry<?, ?> entry) {
+ final EntryEventImpl event;
+ if (region instanceof PartitionedRegion) {
+ event = createEventForPartitionedRegion(sender, cache, region, entry);
+ } else {
+ event = createEventForReplicatedRegion(cache, region, entry);
+ }
+ if (event == null) {
+ return null;
+ }
+ try {
+ return new
GatewaySenderEventImpl(EnumListenerEvent.AFTER_UPDATE_WITH_GENERATE_CALLBACKS,
+ event, null, true);
+ } catch (IOException e) {
+ logger.error("Error when creating event in wan-copy: {}",
e.getMessage());
+ return null;
+ }
+ }
+
+ final CliFunctionResult cancelWanCopyRegion(FunctionContext<Object[]>
context,
+ String regionName, String senderId) {
+ Future<?> execution = executions.remove(getExecutionName(regionName,
senderId));
+ if (execution == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__NO__RUNNING__COMMAND,
+ regionName, senderId));
+ }
+ execution.cancel(true);
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.OK,
+ CliStrings.WAN_COPY_REGION__MSG__EXECUTION__CANCELED);
+ }
+
+ final CliFunctionResult cancelAllWanCopyRegion(FunctionContext<Object[]>
context) {
+ String executionsString = executions.keySet().toString();
+ for (Future<?> execution : executions.values()) {
+ execution.cancel(true);
+ }
+ executions.clear();
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.OK,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__EXECUTIONS__CANCELED,
executionsString));
+ }
+
+ public static String getExecutionName(String regionName, String senderId) {
+ return "(" + regionName + "," + senderId + ")";
+ }
+
+ /**
+ * It runs the actions to be done after a batch has been
+ * sent: throw an interrupted exception if the operation was canceled and
+ * adjust the rate of copying by sleeping if necessary.
+ *
+ * @param startTime time at which the entries started to be copied
+ * @param copiedEntries number of entries copied so far
+ * @param maxRate maximum copying rate
+ */
+ void doPostSendBatchActions(long startTime, int copiedEntries, long maxRate)
+ throws InterruptedException {
+ long sleepMs = getTimeToSleep(startTime, copiedEntries, maxRate);
+ if (sleepMs > 0) {
+ logger.info("{}: Sleeping for {} ms to accommodate to requested maxRate",
+ this.getClass().getSimpleName(), sleepMs);
+ threadSleeper.millis(sleepMs);
+ } else {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException();
+ }
+ }
+ }
+
+ private int getAndIncrementBatchId() {
+ if (batchId + 1 == Integer.MAX_VALUE) {
+ batchId = 0;
+ }
+ return batchId++;
Review comment:
`batchId` is volatile, but this operation is not atomic, which may
introduce a race condition here. Would it be possible to synchronize this
method to eliminate this race?
##########
File path:
geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/WanCopyRegionFunction.java
##########
@@ -0,0 +1,562 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.annotations.VisibleForTesting;
+import org.apache.geode.cache.Declarable;
+import org.apache.geode.cache.EntryDestroyedException;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.AllConnectionsInUseException;
+import org.apache.geode.cache.client.NoAvailableServersException;
+import org.apache.geode.cache.client.ServerConnectivityException;
+import org.apache.geode.cache.client.internal.Connection;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import
org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DefaultEntryEventFactory;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EntrySnapshot;
+import org.apache.geode.internal.cache.EnumListenerEvent;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalRegion;
+import org.apache.geode.internal.cache.NonTXEntry;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.BatchException70;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventDispatcher;
+import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliFunction;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+/**
+ * Class for copying via WAN the contents of a region
+ * It must be executed in all members of the Geode cluster that host the region
+ * to be copied. (called with onServers() or withMembers() passing the list
+ * of all members hosting the region).
+ * It also offers the possibility to cancel an ongoing execution of this
function.
+ * The copying itself is executed in a new thread with a known name
+ * (parameterized with the regionName and senderId) in order to allow
+ * to cancel ongoing invocations by interrupting that thread.
+ *
+ * It accepts the following arguments in an array of objects
+ * 0: regionName (String)
+ * 1: senderId (String)
+ * 2: isCancel (Boolean): If true, it indicates that an ongoing execution of
this
+ * function for the given region and senderId must be stopped. Otherwise,
+ * it indicates that the region must be copied.
+ * 3: maxRate (Long) maximum copy rate in entries per second. In the case of
+ * parallel gateway senders, the maxRate is per server hosting the region.
+ * 4: batchSize (Integer): the size of the batches. Region entries are copied
in batches of the
+ * passed size. After each batch is sent, the function checks if the command
+ * must be canceled and also sleeps for some time if necessary to adjust the
+ * copy rate to the one passed as argument.
+ */
+public class WanCopyRegionFunction extends CliFunction<Object[]> implements
Declarable {
+ private static final Logger logger = LogService.getLogger();
+ private static final long serialVersionUID = 1L;
+
+ public static final String ID = WanCopyRegionFunction.class.getName();
+
+ private static final int MAX_BATCH_SEND_RETRIES = 1;
+
+ private static final int THREAD_POOL_SIZE = 10;
+
+ private final Clock clock;
+ private final ThreadSleeper threadSleeper;
+
+ private static final ExecutorService executor = LoggingExecutors
+ .newFixedThreadPool(THREAD_POOL_SIZE, "wanCopyRegionFunctionThread_",
true);
+
+ /**
+ * Contains the ongoing executions of this function
+ */
+ private static final Map<String, Future<?>> executions = new
ConcurrentHashMap<>();
+
+ private volatile int batchId = 0;
+
+ public WanCopyRegionFunction() {
+ this(Clock.systemDefaultZone(), new ThreadSleeper());
+ }
+
+ @VisibleForTesting
+ WanCopyRegionFunction(Clock clock, ThreadSleeper threadSleeper) {
+ this.clock = clock;
+ this.threadSleeper = threadSleeper;
+ }
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+ @Override
+ public boolean hasResult() {
+ return true;
+ }
+
+ @Override
+ public boolean isHA() {
+ return false;
+ }
+
+ @Override
+ public CliFunctionResult executeFunction(FunctionContext<Object[]> context) {
+ final Object[] args = context.getArguments();
+ if (args.length < 5) {
+ throw new IllegalStateException(
+ "Arguments length does not match required length.");
+ }
+ final String regionName = (String) args[0];
+ final String senderId = (String) args[1];
+ final boolean isCancel = (Boolean) args[2];
+ long maxRate = (Long) args[3];
+ int batchSize = (Integer) args[4];
+
+ if (regionName.endsWith("*") && senderId.equals("*") && isCancel) {
+ return cancelAllWanCopyRegion(context);
+ }
+
+ if (isCancel) {
+ return cancelWanCopyRegion(context, regionName, senderId);
+ }
+
+ final InternalCache cache = (InternalCache) context.getCache();
+
+ final Region<?, ?> region = cache.getRegion(regionName);
+ if (region == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__FOUND,
regionName));
+ }
+
+ GatewaySender sender = cache.getGatewaySender(senderId);
+ if (sender == null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__FOUND,
senderId));
+ }
+
+ if
(!region.getAttributes().getGatewaySenderIds().contains(sender.getId())) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__REGION__NOT__USING_SENDER,
regionName,
+ senderId));
+ }
+
+ if (!sender.isRunning()) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__NOT__RUNNING,
senderId));
+ }
+
+ if (!sender.isParallel() && !((InternalGatewaySender) sender).isPrimary())
{
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.OK,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__SENDER__SERIAL__AND__NOT__PRIMARY,
+ senderId));
+ }
+
+ try {
+ return executeWanCopyRegionFunctionInNewThread(context, region,
regionName, sender, maxRate,
+ batchSize);
+ } catch (InterruptedException | CancellationException e) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+ CliStrings.WAN_COPY_REGION__MSG__CANCELED__BEFORE__HAVING__COPIED);
+ } catch (ExecutionException e) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__EXECUTION__FAILED,
e.getMessage()));
+ }
+ }
+
+ private CliFunctionResult executeWanCopyRegionFunctionInNewThread(
+ FunctionContext<Object[]> context,
+ Region<?, ?> region, String regionName, GatewaySender sender, long
maxRate, int batchSize)
+ throws InterruptedException, ExecutionException, CancellationException {
+ String executionName = getExecutionName(regionName, sender.getId());
+ Callable<CliFunctionResult> callable =
+ new WanCopyRegionCallable(this, context, region, sender, maxRate,
batchSize);
+ FutureTask<CliFunctionResult> future = new FutureTask<>(callable);
+
+ if (executions.putIfAbsent(executionName, future) != null) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+
CliStrings.format(CliStrings.WAN_COPY_REGION__MSG__ALREADY__RUNNING__COMMAND,
+ regionName, sender.getId()));
+ }
+
+ try {
+ executor.execute(future);
+ return future.get();
+ } finally {
+ executions.remove(executionName);
+ }
+ }
+
+ static class WanCopyRegionCallable implements Callable<CliFunctionResult> {
+ private final FunctionContext<Object[]> context;
+ private final Region<?, ?> region;
+ private final GatewaySender sender;
+ private final long maxRate;
+ private final int batchSize;
+ private final WanCopyRegionFunction function;
+
+ public WanCopyRegionCallable(final WanCopyRegionFunction function,
+ final FunctionContext<Object[]> context, final Region<?, ?> region,
+ final GatewaySender sender, final long maxRate,
+ final int batchSize) {
+ this.function = function;
+ this.context = context;
+ this.region = region;
+ this.sender = sender;
+ this.maxRate = maxRate;
+ this.batchSize = batchSize;
+ }
+
+ @Override
+ public CliFunctionResult call() throws Exception {
+ return function.wanCopyRegion(context, region, sender, maxRate,
batchSize);
+ }
+ }
+
+ @VisibleForTesting
+ CliFunctionResult wanCopyRegion(FunctionContext<Object[]> context, Region<?,
?> region,
+ GatewaySender sender, long maxRate, int batchSize) throws
InterruptedException {
+ ConnectionState connectionState = new ConnectionState();
+ int copiedEntries = 0;
+ final InternalCache cache = (InternalCache) context.getCache();
+ Iterator<?> entriesIter = getEntries(region, sender).iterator();
+ final long startTime = clock.millis();
+
+ try {
+ while (entriesIter.hasNext()) {
+ List<GatewayQueueEvent<?, ?>> batch =
+ createBatch((InternalRegion) region, sender, batchSize, cache,
entriesIter);
+ if (batch.size() == 0) {
+ continue;
+ }
+ Optional<CliFunctionResult> connectionError =
+ connectionState.connectIfNeeded(context, sender);
+ if (connectionError.isPresent()) {
+ return connectionError.get();
+ }
+ Optional<CliFunctionResult> error =
+ sendBatch(context, sender, batch, connectionState, copiedEntries);
+ if (error.isPresent()) {
+ return error.get();
+ }
+ copiedEntries += batch.size();
+ doPostSendBatchActions(startTime, copiedEntries, maxRate);
+ }
+ } finally {
+ connectionState.close();
+ }
+
+ if (region.isDestroyed()) {
+ return new CliFunctionResult(context.getMemberName(),
CliFunctionResult.StatusState.ERROR,
+ CliStrings.format(
+ CliStrings.WAN_COPY_REGION__MSG__ERROR__AFTER__HAVING__COPIED,
+ "Region destroyed",
+ copiedEntries));
+ }
Review comment:
This case has no test coverage. It would be good to add Unit and
possibly DUnit tests to confirm that this message is returned when expected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
