You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by mi...@apache.org on 2021/04/08 08:46:29 UTC
[geode] branch develop updated: GEODE-8605: Alter Gateway Sender
command (#5630)
This is an automated email from the ASF dual-hosted git repository.
mivanac pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 7aa0382 GEODE-8605: Alter Gateway Sender command (#5630)
7aa0382 is described below
commit 7aa03824cc552433f034c8b27f4241d96bdb6383
Author: Mario Ivanac <48...@users.noreply.github.com>
AuthorDate: Thu Apr 8 10:45:31 2021 +0200
GEODE-8605: Alter Gateway Sender command (#5630)
* GEODE-8605: Alter Gateway Sender command
* GEODE-8605: fix for failing integration test
* GEODE-8605: added distributed test
* GEODE-8605: update stability of distributed tests
* GEODE-8605: added documentation impacts
* GEODE-8605: added test for new function
* GEODE-8605: added filter test
* GEODE-8605: update after comments
* GEODE-8605: update solution
* GEODE-8605: add test for parallel gw sender
* GEODE-8605: added setters instead of update
* GEODE-8605: added ParallelGatewaySender tests
* GEODE-8605: added test for SerialGatewaySender
* GEODE-8605: update solution
* GEODE-8605: add option to remove gateway-event-filters
* GEODE-8605: update of remove filters solution
* GEODE-8605: rebase solution
* GEODE-8605: added serialVersionUID to GatewaySender
* GEODE-8605: added test case
* GEODE-8605: remove changes in CacheConfig.GatewaySender
* GEODE-8605: add setters to GatewaySenderAttributes
* GEODE-8605: small update
* GEODE-8605: update after rebase
* GEODE-8605: update after comments
* GEODE-8605: update after comments without atomic change
---
.../SerialAsyncEventQueueImplJUnitTest.java | 22 +-
.../internal/AsyncEventQueueFactoryImpl.java | 76 +-
.../geode/cache/configuration/CacheConfig.java | 4 +
.../org/apache/geode/cache/wan/GatewaySender.java | 43 +-
.../internal/cache/wan/AbstractGatewaySender.java | 63 +-
.../wan/AbstractGatewaySenderEventProcessor.java | 9 +-
.../cache/wan/GatewaySenderAttributes.java | 157 +++-
.../cache/xmlcache/AsyncEventQueueCreation.java | 28 +-
.../geode/management/internal/i18n/CliStrings.java | 48 ++
.../internal/ParallelAsyncEventQueueImplTest.java | 4 +-
.../gfsh/command-pages/alter.html.md.erb | 79 ++
.../cli/commands/AlterGatewaySenderCommand.java | 276 +++++++
.../cli/commands/CommandAvailabilityIndicator.java | 2 +-
.../cli/functions/AlterGatewaySenderFunction.java | 122 +++
.../cli/functions/GatewaySenderFunctionArgs.java | 18 +-
.../sanctioned-geode-gfsh-serializables.txt | 1 +
.../commands/AlterGatewaySenderCommandTest.java | 138 ++++
.../commands/CreateGatewaySenderCommandTest.java | 2 +-
.../functions/AlterGatewaySenderFunctionTest.java | 122 +++
.../ParallelGatewaySenderOperationsDUnitTest.java | 477 +++++++++++
...erialGatewaySenderAlterOperationsDUnitTest.java | 885 +++++++++++++++++++++
.../AlterGatewaySenderCommandDUnitTest.java | 488 ++++++++++++
.../wan/internal/GatewaySenderFactoryImpl.java | 104 +--
.../parallel/ParallelGatewaySenderImplTest.java | 6 +-
24 files changed, 3012 insertions(+), 162 deletions(-)
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImplJUnitTest.java b/geode-core/src/integrationTest/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImplJUnitTest.java
index 9138c8c..1adae4f 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImplJUnitTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImplJUnitTest.java
@@ -34,19 +34,30 @@ public class SerialAsyncEventQueueImplJUnitTest {
@Before
public void setUp() {
- CacheFactory cf = new CacheFactory().set(MCAST_PORT, "0");
- cache = (InternalCache) cf.create();
+ try {
+ cache = (InternalCache) CacheFactory.getAnyInstance();
+ } catch (Exception e) {
+ // ignore
+ }
+ if (null == cache) {
+ cache = (InternalCache) new CacheFactory().set(MCAST_PORT, "0").create();
+ }
}
@After
public void tearDown() {
- cache.close();
+ if (cache != null && !cache.isClosed()) {
+ cache.close();
+ cache = null;
+ }
}
@Test
public void testStopClearsStats() {
+
GatewaySenderAttributes attrs = new GatewaySenderAttributes();
- attrs.id = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
+ String tempId = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
+ attrs.setId(tempId);
SerialAsyncEventQueueImpl queue = new SerialAsyncEventQueueImpl(cache,
cache.getInternalDistributedSystem().getStatisticsManager(), cache.getStatisticsClock(),
attrs);
@@ -71,7 +82,8 @@ public class SerialAsyncEventQueueImplJUnitTest {
@Test
public void testStopStart() {
GatewaySenderAttributes attrs = new GatewaySenderAttributes();
- attrs.id = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
+ String tempId = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
+ attrs.setId(tempId);
SerialAsyncEventQueueImpl queue = new SerialAsyncEventQueueImpl(cache,
cache.getInternalDistributedSystem().getStatisticsManager(), cache.getStatisticsClock(),
attrs);
diff --git a/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java b/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
index 004cb4e..700cc4b 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java
@@ -64,60 +64,60 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
this.cache = cache;
this.gatewaySenderAttributes = gatewaySenderAttributes;
// set a different default for batchTimeInterval for AsyncEventQueue
- this.gatewaySenderAttributes.batchTimeInterval = batchTimeInterval;
+ this.gatewaySenderAttributes.setBatchTimeInterval(batchTimeInterval);
}
@Override
public AsyncEventQueueFactory setBatchSize(int size) {
- gatewaySenderAttributes.batchSize = size;
+ gatewaySenderAttributes.setBatchSize(size);
return this;
}
@Override
public AsyncEventQueueFactory setPersistent(boolean isPersistent) {
- gatewaySenderAttributes.isPersistenceEnabled = isPersistent;
+ gatewaySenderAttributes.setPersistenceEnabled(isPersistent);
return this;
}
@Override
public AsyncEventQueueFactory setDiskStoreName(String name) {
- gatewaySenderAttributes.diskStoreName = name;
+ gatewaySenderAttributes.setDiskStoreName(name);
return this;
}
@Override
public AsyncEventQueueFactory setMaximumQueueMemory(int memory) {
- gatewaySenderAttributes.maximumQueueMemory = memory;
+ gatewaySenderAttributes.setMaximumQueueMemory(memory);
return this;
}
@Override
public AsyncEventQueueFactory setDiskSynchronous(boolean isSynchronous) {
- gatewaySenderAttributes.isDiskSynchronous = isSynchronous;
+ gatewaySenderAttributes.setDiskSynchronous(isSynchronous);
return this;
}
@Override
public AsyncEventQueueFactory setBatchTimeInterval(int batchTimeInterval) {
- gatewaySenderAttributes.batchTimeInterval = batchTimeInterval;
+ gatewaySenderAttributes.setBatchTimeInterval(batchTimeInterval);
return this;
}
@Override
public AsyncEventQueueFactory setBatchConflationEnabled(boolean isConflation) {
- gatewaySenderAttributes.isBatchConflationEnabled = isConflation;
+ gatewaySenderAttributes.setBatchConflationEnabled(isConflation);
return this;
}
@Override
public AsyncEventQueueFactory setDispatcherThreads(int numThreads) {
- gatewaySenderAttributes.dispatcherThreads = numThreads;
+ gatewaySenderAttributes.setDispatcherThreads(numThreads);
return this;
}
@Override
public AsyncEventQueueFactory setOrderPolicy(OrderPolicy policy) {
- gatewaySenderAttributes.policy = policy;
+ gatewaySenderAttributes.setOrderPolicy(policy);
return this;
}
@@ -129,14 +129,14 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
@Override
public AsyncEventQueueFactory removeGatewayEventFilter(GatewayEventFilter filter) {
- gatewaySenderAttributes.eventFilters.remove(filter);
+ gatewaySenderAttributes.getGatewayEventFilters().remove(filter);
return this;
}
@Override
public AsyncEventQueueFactory setGatewayEventSubstitutionListener(
GatewayEventSubstitutionFilter filter) {
- gatewaySenderAttributes.eventSubstitutionFilter = filter;
+ gatewaySenderAttributes.setEventSubstitutionFilter(filter);
return this;
}
@@ -192,7 +192,7 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
}
private GatewaySender create(String id) {
- gatewaySenderAttributes.id = id;
+ gatewaySenderAttributes.setId(id);
if (gatewaySenderAttributes.getDispatcherThreads() <= 0) {
throw new AsyncEventQueueConfigurationException(
@@ -222,7 +222,7 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
} else {
if (gatewaySenderAttributes.getOrderPolicy() == null
&& gatewaySenderAttributes.getDispatcherThreads() > 1) {
- gatewaySenderAttributes.policy = GatewaySender.DEFAULT_ORDER_POLICY;
+ gatewaySenderAttributes.setOrderPolicy(GatewaySender.DEFAULT_ORDER_POLICY);
}
if (cache instanceof CacheCreation) {
@@ -238,46 +238,48 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
}
public void configureAsyncEventQueue(AsyncEventQueue asyncQueueCreation) {
- gatewaySenderAttributes.batchSize = asyncQueueCreation.getBatchSize();
- gatewaySenderAttributes.batchTimeInterval = asyncQueueCreation.getBatchTimeInterval();
- gatewaySenderAttributes.isBatchConflationEnabled =
- asyncQueueCreation.isBatchConflationEnabled();
- gatewaySenderAttributes.isPersistenceEnabled = asyncQueueCreation.isPersistent();
- gatewaySenderAttributes.diskStoreName = asyncQueueCreation.getDiskStoreName();
- gatewaySenderAttributes.isDiskSynchronous = asyncQueueCreation.isDiskSynchronous();
- gatewaySenderAttributes.maximumQueueMemory = asyncQueueCreation.getMaximumQueueMemory();
- gatewaySenderAttributes.isParallel = asyncQueueCreation.isParallel();
- gatewaySenderAttributes.isBucketSorted =
- ((AsyncEventQueueCreation) asyncQueueCreation).isBucketSorted();
- gatewaySenderAttributes.dispatcherThreads = asyncQueueCreation.getDispatcherThreads();
- gatewaySenderAttributes.policy = asyncQueueCreation.getOrderPolicy();
- gatewaySenderAttributes.eventFilters = asyncQueueCreation.getGatewayEventFilters();
- gatewaySenderAttributes.eventSubstitutionFilter =
- asyncQueueCreation.getGatewayEventSubstitutionFilter();
- gatewaySenderAttributes.isForInternalUse = true;
- gatewaySenderAttributes.forwardExpirationDestroy =
- asyncQueueCreation.isForwardExpirationDestroy();
+ gatewaySenderAttributes.setBatchSize(asyncQueueCreation.getBatchSize());
+ gatewaySenderAttributes.setBatchTimeInterval(asyncQueueCreation.getBatchTimeInterval());
+ gatewaySenderAttributes
+ .setBatchConflationEnabled(asyncQueueCreation.isBatchConflationEnabled());
+ gatewaySenderAttributes.setPersistenceEnabled(asyncQueueCreation.isPersistent());
+ gatewaySenderAttributes.setDiskStoreName(asyncQueueCreation.getDiskStoreName());
+ gatewaySenderAttributes.setDiskSynchronous(asyncQueueCreation.isDiskSynchronous());
+ gatewaySenderAttributes.setMaximumQueueMemory(asyncQueueCreation.getMaximumQueueMemory());
+ gatewaySenderAttributes.setParallel(asyncQueueCreation.isParallel());
+ gatewaySenderAttributes
+ .setBucketSorted(((AsyncEventQueueCreation) asyncQueueCreation).isBucketSorted());
+ gatewaySenderAttributes.setDispatcherThreads(asyncQueueCreation.getDispatcherThreads());
+ gatewaySenderAttributes.setOrderPolicy(asyncQueueCreation.getOrderPolicy());
+ for (GatewayEventFilter filter : asyncQueueCreation.getGatewayEventFilters()) {
+ gatewaySenderAttributes.getGatewayEventFilters().add(filter);
+ }
+ gatewaySenderAttributes
+ .setEventSubstitutionFilter(asyncQueueCreation.getGatewayEventSubstitutionFilter());
+ gatewaySenderAttributes.setForInternalUse(true);
+ gatewaySenderAttributes
+ .setForwardExpirationDestroy(asyncQueueCreation.isForwardExpirationDestroy());
}
@Override
public AsyncEventQueueFactory setParallel(boolean isParallel) {
- gatewaySenderAttributes.isParallel = isParallel;
+ gatewaySenderAttributes.setParallel(isParallel);
return this;
}
public AsyncEventQueueFactory setBucketSorted(boolean isbucketSorted) {
- gatewaySenderAttributes.isBucketSorted = isbucketSorted;
+ gatewaySenderAttributes.setBucketSorted(isbucketSorted);
return this;
}
public AsyncEventQueueFactory setIsMetaQueue(boolean isMetaQueue) {
- gatewaySenderAttributes.isMetaQueue = isMetaQueue;
+ gatewaySenderAttributes.setMetaQueue(isMetaQueue);
return this;
}
@Override
public AsyncEventQueueFactory setForwardExpirationDestroy(boolean forward) {
- gatewaySenderAttributes.forwardExpirationDestroy = forward;
+ gatewaySenderAttributes.setForwardExpirationDestroy(forward);
return this;
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/configuration/CacheConfig.java b/geode-core/src/main/java/org/apache/geode/cache/configuration/CacheConfig.java
index 85fe90e..a4beb25 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/configuration/CacheConfig.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/configuration/CacheConfig.java
@@ -2689,6 +2689,10 @@ public class CacheConfig {
return this.gatewayEventFilters;
}
+ public boolean areGatewayEventFiltersUpdated() {
+ return gatewayEventFilters != null;
+ }
+
/**
* Gets the value of the gatewayEventSubstitutionFilter property.
*
diff --git a/geode-core/src/main/java/org/apache/geode/cache/wan/GatewaySender.java b/geode-core/src/main/java/org/apache/geode/cache/wan/GatewaySender.java
index 938c177..8e498c3 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/wan/GatewaySender.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/wan/GatewaySender.java
@@ -462,7 +462,6 @@ public interface GatewaySender {
int getMaxParallelismForReplicatedRegion();
-
/**
* Destroys the GatewaySender.
* <p>
@@ -481,4 +480,46 @@ public interface GatewaySender {
*
*/
boolean getEnforceThreadsConnectSameReceiver();
+
+
+ /**
+ * Set AlertThreshold for this GatewaySender.
+ *
+ * @since Geode 1.14
+ *
+ */
+ void setAlertThreshold(int alertThreshold);
+
+ /**
+ * Set BatchSize for this GatewaySender.
+ *
+ * @since Geode 1.14
+ *
+ */
+ void setBatchSize(int batchSize);
+
+ /**
+ * Set BatchTimeInterval for this GatewaySender.
+ *
+ * @since Geode 1.14
+ *
+ */
+ void setBatchTimeInterval(int batchTimeInterval);
+
+ /**
+ * Set GroupTransactionEvents for this GatewaySender.
+ *
+ * @since Geode 1.14
+ *
+ */
+ void setGroupTransactionEvents(boolean groupTransactionEvents);
+
+ /**
+ * Set GatewayEventFilters for this GatewaySender.
+ *
+ * @since Geode 1.14
+ *
+ */
+ void setGatewayEventFilters(List<GatewayEventFilter> filters);
+
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
index 2f69893..6760727 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
@@ -18,6 +18,7 @@ import static org.apache.geode.internal.statistics.StatisticsClockFactory.disabl
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -119,7 +120,7 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
protected boolean isPersistence;
- protected int alertThreshold;
+ protected volatile int alertThreshold;
protected boolean manualStart;
@@ -135,9 +136,9 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
protected String diskStoreName;
- protected List<GatewayEventFilter> eventFilters;
+ protected volatile List<GatewayEventFilter> eventFilters;
- protected List<GatewayTransportFilter> transFilters;
+ protected volatile List<GatewayTransportFilter> transFilters;
protected List<AsyncEventListener> listeners;
@@ -261,8 +262,8 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
this.isForInternalUse = attrs.isForInternalUse();
this.diskStoreName = attrs.getDiskStoreName();
this.remoteDSId = attrs.getRemoteDSId();
- this.eventFilters = attrs.getGatewayEventFilters();
- this.transFilters = attrs.getGatewayTransportFilters();
+ this.eventFilters = Collections.unmodifiableList(attrs.getGatewayEventFilters());
+ this.transFilters = Collections.unmodifiableList(attrs.getGatewayTransportFilters());
this.listeners = attrs.getAsyncEventListeners();
this.substitutionFilter = attrs.getGatewayEventSubstitutionFilter();
this.locatorDiscoveryCallback = attrs.getGatewayLocatoDiscoveryCallback();
@@ -535,19 +536,26 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
@Override
public void removeGatewayEventFilter(GatewayEventFilter filter) {
- this.eventFilters.remove(filter);
+ if (filter == null) {
+ return;
+ }
+ if (this.eventFilters.isEmpty()) {
+ return;
+ }
+ List<GatewayEventFilter> templist = new ArrayList<>(this.eventFilters);
+ templist.remove(filter);
+ this.eventFilters = Collections.unmodifiableList(templist);
}
@Override
public void addGatewayEventFilter(GatewayEventFilter filter) {
- if (this.eventFilters.isEmpty()) {
- this.eventFilters = new ArrayList<GatewayEventFilter>();
- }
if (filter == null) {
throw new IllegalStateException(
"null value can not be added to gateway-event-filters list");
}
- this.eventFilters.add(filter);
+ List<GatewayEventFilter> templist = new ArrayList<>(this.eventFilters);
+ templist.add(filter);
+ this.eventFilters = Collections.unmodifiableList(templist);
}
@Override
@@ -672,6 +680,41 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
"GatewaySender {} has been rebalanced", this);
}
+ @Override
+ public void setAlertThreshold(int alertThreshold) {
+ this.alertThreshold = alertThreshold;
+ };
+
+ @Override
+ public void setBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ if (this.eventProcessor != null) {
+ this.eventProcessor.setBatchSize(this.batchSize);
+ }
+ };
+
+ @Override
+ public void setBatchTimeInterval(int batchTimeInterval) {
+ this.batchTimeInterval = batchTimeInterval;
+ if (this.eventProcessor != null) {
+ this.eventProcessor.setBatchTimeInterval(this.batchTimeInterval);
+ }
+ };
+
+ @Override
+ public void setGroupTransactionEvents(boolean groupTransactionEvents) {
+ this.groupTransactionEvents = groupTransactionEvents;
+ };
+
+ @Override
+ public void setGatewayEventFilters(List<GatewayEventFilter> filters) {
+ if (filters.isEmpty()) {
+ this.eventFilters = Collections.emptyList();
+ } else {
+ this.eventFilters = Collections.unmodifiableList(filters);
+ }
+ };
+
public boolean beforeEnqueue(GatewayQueueEvent gatewayEvent) {
boolean enqueue = true;
for (GatewayEventFilter filter : getGatewayEventFilters()) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
index 1361421..ba271d8 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
@@ -146,12 +146,14 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
* occurs.
*/
private int batchSize;
+ private int batchTimeInterval;
public AbstractGatewaySenderEventProcessor(String string,
GatewaySender sender, ThreadsMonitoring tMonitoring) {
super(string);
this.sender = (AbstractGatewaySender) sender;
this.batchSize = sender.getBatchSize();
+ this.batchTimeInterval = sender.getBatchTimeInterval();
this.threadMonitoring = tMonitoring;
}
@@ -262,6 +264,10 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
}
}
+ protected void setBatchTimeInterval(int batchTimeInterval) {
+ this.batchTimeInterval = batchTimeInterval;
+ }
+
/**
* Returns the current batch id to be used to identify the next batch.
*
@@ -433,7 +439,6 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
final boolean isDebugEnabled = logger.isDebugEnabled();
final boolean isTraceEnabled = logger.isTraceEnabled();
- final int batchTimeInterval = sender.getBatchTimeInterval();
final GatewaySenderStats statistics = this.sender.getStatistics();
if (isDebugEnabled) {
@@ -515,7 +520,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
* Thread.currentThread().interrupt(); } } }
*/
}
- events = this.queue.peek(this.batchSize, batchTimeInterval);
+ events = this.queue.peek(this.batchSize, this.batchTimeInterval);
} catch (InterruptedException e) {
interrupted = true;
this.sender.getCancelCriterion().checkCancelInProgress(e);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAttributes.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAttributes.java
index 581b576..012fde6 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAttributes.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAttributes.java
@@ -31,63 +31,163 @@ public class GatewaySenderAttributes {
public static final boolean DEFAULT_IS_META_QUEUE = false;
- public int socketBufferSize = GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE;
+ private int socketBufferSize = GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE;
- public int socketReadTimeout = GatewaySender.DEFAULT_SOCKET_READ_TIMEOUT;
+ private int socketReadTimeout = GatewaySender.DEFAULT_SOCKET_READ_TIMEOUT;
- public int maximumQueueMemory = GatewaySender.DEFAULT_MAXIMUM_QUEUE_MEMORY;
+ private int maximumQueueMemory = GatewaySender.DEFAULT_MAXIMUM_QUEUE_MEMORY;
- public int batchSize = GatewaySender.DEFAULT_BATCH_SIZE;
+ private int batchSize = GatewaySender.DEFAULT_BATCH_SIZE;
- public int batchTimeInterval = GatewaySender.DEFAULT_BATCH_TIME_INTERVAL;
+ private int batchTimeInterval = GatewaySender.DEFAULT_BATCH_TIME_INTERVAL;
- public boolean isBatchConflationEnabled = GatewaySender.DEFAULT_BATCH_CONFLATION;
+ private boolean isBatchConflationEnabled = GatewaySender.DEFAULT_BATCH_CONFLATION;
- public boolean isPersistenceEnabled = GatewaySender.DEFAULT_PERSISTENCE_ENABLED;
+ private boolean isPersistenceEnabled = GatewaySender.DEFAULT_PERSISTENCE_ENABLED;
- public int alertThreshold = GatewaySender.DEFAULT_ALERT_THRESHOLD;
+ private int alertThreshold = GatewaySender.DEFAULT_ALERT_THRESHOLD;
- public boolean manualStart = GatewaySender.DEFAULT_MANUAL_START;
+ private boolean manualStart = GatewaySender.DEFAULT_MANUAL_START;
- public String diskStoreName;
+ private String diskStoreName;
- public List<GatewayEventFilter> eventFilters = new ArrayList<GatewayEventFilter>();
+ private List<GatewayEventFilter> eventFilters = new ArrayList<GatewayEventFilter>();
- public ArrayList<GatewayTransportFilter> transFilters = new ArrayList<GatewayTransportFilter>();
+ private ArrayList<GatewayTransportFilter> transFilters = new ArrayList<GatewayTransportFilter>();
- public List<AsyncEventListener> listeners = new ArrayList<AsyncEventListener>();
+ private List<AsyncEventListener> listeners = new ArrayList<AsyncEventListener>();
- public GatewayEventSubstitutionFilter eventSubstitutionFilter;
+ private GatewayEventSubstitutionFilter eventSubstitutionFilter;
- public String id;
+ private String id;
- public int remoteDs = GatewaySender.DEFAULT_DISTRIBUTED_SYSTEM_ID;
+ private int remoteDs = GatewaySender.DEFAULT_DISTRIBUTED_SYSTEM_ID;
- public LocatorDiscoveryCallback locatorDiscoveryCallback;
+ private LocatorDiscoveryCallback locatorDiscoveryCallback;
- public boolean isDiskSynchronous = GatewaySender.DEFAULT_DISK_SYNCHRONOUS;
+ private boolean isDiskSynchronous = GatewaySender.DEFAULT_DISK_SYNCHRONOUS;
- public OrderPolicy policy;
+ private OrderPolicy policy;
- public int dispatcherThreads = GatewaySender.DEFAULT_DISPATCHER_THREADS;
+ private int dispatcherThreads = GatewaySender.DEFAULT_DISPATCHER_THREADS;
- public int parallelism = GatewaySender.DEFAULT_PARALLELISM_REPLICATED_REGION;
+ private int parallelism = GatewaySender.DEFAULT_PARALLELISM_REPLICATED_REGION;
- public boolean isParallel = GatewaySender.DEFAULT_IS_PARALLEL;
+ private boolean isParallel = GatewaySender.DEFAULT_IS_PARALLEL;
- public boolean groupTransactionEvents = GatewaySender.DEFAULT_MUST_GROUP_TRANSACTION_EVENTS;
+ private boolean groupTransactionEvents = GatewaySender.DEFAULT_MUST_GROUP_TRANSACTION_EVENTS;
- public boolean isForInternalUse = GatewaySender.DEFAULT_IS_FOR_INTERNAL_USE;
+ private boolean isForInternalUse = GatewaySender.DEFAULT_IS_FOR_INTERNAL_USE;
- public boolean isBucketSorted = GatewaySenderAttributes.DEFAULT_IS_BUCKETSORTED;
+ private boolean isBucketSorted = GatewaySenderAttributes.DEFAULT_IS_BUCKETSORTED;
- public boolean isMetaQueue = GatewaySenderAttributes.DEFAULT_IS_META_QUEUE;
+ private boolean isMetaQueue = GatewaySenderAttributes.DEFAULT_IS_META_QUEUE;
- public boolean forwardExpirationDestroy = GatewaySender.DEFAULT_FORWARD_EXPIRATION_DESTROY;
+ private boolean forwardExpirationDestroy = GatewaySender.DEFAULT_FORWARD_EXPIRATION_DESTROY;
- public boolean enforceThreadsConnectSameReceiver =
+ private boolean enforceThreadsConnectSameReceiver =
GatewaySender.DEFAULT_ENFORCE_THREADS_CONNECT_SAME_RECEIVER;
+ public void setSocketBufferSize(int bufferSize) {
+ socketBufferSize = bufferSize;
+ }
+
+ public void setSocketReadTimeout(int readTimeout) {
+ socketReadTimeout = readTimeout;
+ }
+
+ public void setMaximumQueueMemory(int maxQueueMemory) {
+ maximumQueueMemory = maxQueueMemory;
+ }
+
+ public void setBatchSize(int batchsize) {
+ batchSize = batchsize;
+ }
+
+ public void setBatchTimeInterval(int batchtimeinterval) {
+ batchTimeInterval = batchtimeinterval;
+ }
+
+ public void setBatchConflationEnabled(boolean batchConfEnabled) {
+ isBatchConflationEnabled = batchConfEnabled;
+ }
+
+ public void setPersistenceEnabled(boolean persistenceEnabled) {
+ isPersistenceEnabled = persistenceEnabled;
+ }
+
+ public void setAlertThreshold(int alertThresh) {
+ alertThreshold = alertThresh;
+ }
+
+ public void setManualStart(boolean manualstart) {
+ manualStart = manualstart;
+ }
+
+ public void setDiskStoreName(String diskstorename) {
+ diskStoreName = diskstorename;
+ }
+
+ public void setEventSubstitutionFilter(GatewayEventSubstitutionFilter eventsubstitutionfilter) {
+ eventSubstitutionFilter = eventsubstitutionfilter;
+ }
+
+ public void setId(String idString) {
+ id = idString;
+ }
+
+ public void setRemoteDs(int rDs) {
+ remoteDs = rDs;
+ }
+
+ public void setLocatorDiscoveryCallback(LocatorDiscoveryCallback locatorDiscCall) {
+ locatorDiscoveryCallback = locatorDiscCall;
+ }
+
+ public void setDiskSynchronous(boolean diskSynchronous) {
+ isDiskSynchronous = diskSynchronous;
+ }
+
+ public void setOrderPolicy(OrderPolicy orderpolicy) {
+ policy = orderpolicy;
+ }
+
+ public void setDispatcherThreads(int dispatchThreads) {
+ dispatcherThreads = dispatchThreads;
+ }
+
+ public void setParallelism(int tempParallelism) {
+ parallelism = tempParallelism;
+ }
+
+ public void setParallel(boolean parallel) {
+ isParallel = parallel;
+ }
+
+ public void setGroupTransactionEvents(boolean groupTransEvents) {
+ groupTransactionEvents = groupTransEvents;
+ }
+
+ public void setForInternalUse(boolean forInternalUse) {
+ isForInternalUse = forInternalUse;
+ }
+
+ public void setBucketSorted(boolean bucketSorted) {
+ isBucketSorted = bucketSorted;
+ }
+
+ public void setMetaQueue(boolean metaQueue) {
+ isMetaQueue = metaQueue;
+ }
+
+ public void setForwardExpirationDestroy(boolean forwardexpirationdestroy) {
+ forwardExpirationDestroy = forwardexpirationdestroy;
+ }
+
+ public void setEnforceThreadsConnectSameReceiver(boolean enforcethreadsconnectsamereceiver) {
+ enforceThreadsConnectSameReceiver = enforcethreadsconnectsamereceiver;
+ }
+
public int getSocketBufferSize() {
return this.socketBufferSize;
}
@@ -211,4 +311,5 @@ public class GatewaySenderAttributes {
public boolean getEnforceThreadsConnectSameReceiver() {
return this.enforceThreadsConnectSameReceiver;
}
+
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/AsyncEventQueueCreation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/AsyncEventQueueCreation.java
index eb29f01..0785163 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/AsyncEventQueueCreation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/AsyncEventQueueCreation.java
@@ -49,21 +49,21 @@ public class AsyncEventQueueCreation implements AsyncEventQueue {
public AsyncEventQueueCreation(String id, GatewaySenderAttributes senderAttrs,
AsyncEventListener eventListener) {
this.id = id;
- this.batchSize = senderAttrs.batchSize;
- this.batchTimeInterval = senderAttrs.batchTimeInterval;
- this.isBatchConflationEnabled = senderAttrs.isBatchConflationEnabled;
- this.isPersistent = senderAttrs.isPersistenceEnabled;
- this.diskStoreName = senderAttrs.diskStoreName;
- this.isDiskSynchronous = senderAttrs.isDiskSynchronous;
- this.maxQueueMemory = senderAttrs.maximumQueueMemory;
- this.isParallel = senderAttrs.isParallel;
- this.dispatcherThreads = senderAttrs.dispatcherThreads;
- this.orderPolicy = senderAttrs.policy;
+ this.batchSize = senderAttrs.getBatchSize();
+ this.batchTimeInterval = senderAttrs.getBatchTimeInterval();
+ this.isBatchConflationEnabled = senderAttrs.isBatchConflationEnabled();
+ this.isPersistent = senderAttrs.isPersistenceEnabled();
+ this.diskStoreName = senderAttrs.getDiskStoreName();
+ this.isDiskSynchronous = senderAttrs.isDiskSynchronous();
+ this.maxQueueMemory = senderAttrs.getMaximumQueueMemory();
+ this.isParallel = senderAttrs.isParallel();
+ this.dispatcherThreads = senderAttrs.getDispatcherThreads();
+ this.orderPolicy = senderAttrs.getOrderPolicy();
this.asyncEventListener = eventListener;
- this.isBucketSorted = senderAttrs.isBucketSorted;
- this.gatewayEventFilters = senderAttrs.eventFilters;
- this.gatewayEventSubstitutionFilter = senderAttrs.eventSubstitutionFilter;
- this.forwardExpirationDestroy = senderAttrs.forwardExpirationDestroy;
+ this.isBucketSorted = senderAttrs.isBucketSorted();
+ this.gatewayEventFilters = senderAttrs.getGatewayEventFilters();
+ this.gatewayEventSubstitutionFilter = senderAttrs.getGatewayEventSubstitutionFilter();
+ this.forwardExpirationDestroy = senderAttrs.isForwardExpirationDestroy();
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/i18n/CliStrings.java b/geode-core/src/main/java/org/apache/geode/management/internal/i18n/CliStrings.java
index 20966ae..e25fe77 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/i18n/CliStrings.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/i18n/CliStrings.java
@@ -85,6 +85,8 @@ public class CliStrings {
public static final String JAR = "jar";
public static final String JARS = "jars";
+ public static final String NULL = "null";
+
public static final String IFEXISTS_HELP =
"If true, the command will be a no-op if the entity does not exist.";
public static final String IFNOTEXISTS_HELP =
@@ -324,6 +326,50 @@ public class CliStrings {
public static final String ALTER_DISK_STORE__REMOVE__HELP =
"Whether to remove the region from the disk store.";
+ /* alter gateway-sender */
+ public static final String ALTER_GATEWAYSENDER = "alter gateway-sender";
+ public static final String ALTER_GATEWAYSENDER__HELP =
+ "Alter some options for the Gateway Sender on a member or members.";
+ public static final String ALTER_GATEWAYSENDER__ID = "id";
+ public static final String ALTER_GATEWAYSENDER__ID__HELP = "ID of the gateway sender.";
+
+ public static final String ALTER_GATEWAYSENDER__GROUP__HELP =
+ "Group(s) of members on which to alter the gateway sender option.";
+ public static final String ALTER_GATEWAYSENDER__MEMBER__HELP =
+ "Name/Id of the member on which to alter the gateway sender option.";
+
+ public static final String ALTER_GATEWAYSENDER__ALERTTHRESHOLD = "alert-threshold";
+ public static final String ALTER_GATEWAYSENDER__ALERTTHRESHOLD__HELP =
+ "The alert threshold for entries in a gateway sender's queue.";
+
+ public static final String ALTER_GATEWAYSENDER__BATCHSIZE = "batch-size";
+ public static final String ALTER_GATEWAYSENDER__BATCHSIZE__HELP =
+ "The batch size for the gateway sender.";
+
+ public static final String ALTER_GATEWAYSENDER__BATCHTIMEINTERVAL = "batch-time-interval";
+ public static final String ALTER_GATEWAYSENDER__BATCHTIMEINTERVAL__HELP =
+ "The batch time interval for the gateway sender.";
+
+ public static final String ALTER_GATEWAYSENDER__GATEWAYEVENTFILTER = "gateway-event-filter";
+ public static final String ALTER_GATEWAYSENDER__GATEWAYEVENTFILTER__HELP =
+ "The list of fully qualified class names of GatewayEventFilters (separated by commas) to be associated with the GatewaySender.\n"
+ + "This serves as a callback for users to filter out events before dispatching to the remote distributed system.\n"
+ + "E.g gateway-event-filter=com.user.filters.MyFilter1,com.user.filters.MyFilters2";
+
+ public static final String ALTER_GATEWAYSENDER__GROUPTRANSACTIONEVENTS =
+ "group-transaction-events";
+ public static final String ALTER_GATEWAYSENDER__GROUPTRANSACTIONEVENTS__HELP =
+ "Ensure that all the events of a transaction are sent in the same batch, i.e., they are never spread across different batches.\n"
+ + "Only allowed on serial gateway senders with 1 dispatcher thread or on parallel ones.\n"
+ + "Note that in order to work for a transaction, the regions to which the transaction \n"
+ + "events belong must be replicated by the same set of senders with this flag enabled.";
+
+ public static final String ALTER_GATEWAYSENDER__RELEVANT__OPTION__MESSAGE =
+ "Please provide a relevant parameter(s)";
+ public static final String ALTER_GATEWAYSENDER__MSG__CAN_NOT_CREATE_DIFFERENT_VERSIONS =
+ "Alter Gateway Sender cannot be performed until all members are the current version";
+
+
/* 'alter region' command */
public static final String ALTER_REGION = "alter region";
public static final String ALTER_REGION__HELP =
@@ -2979,6 +3025,8 @@ public class CliStrings {
"Could not invoke start gateway sender {0} operation on members due to {1}";
public static final String GATEWAY_SENDER_0_COULD_NOT_BE_STARTED_ON_MEMBER_DUE_TO_1 =
"Could not start gateway sender {0} on member due to {1}";
+ public static final String GATEWAY_SENDER_0_IS_UPDATED_ON_MEMBER_1 =
+ "GatewaySender {0} is updated on member {1}";
/* end gateway command messages */
/***
diff --git a/geode-core/src/test/java/org/apache/geode/cache/asyncqueue/internal/ParallelAsyncEventQueueImplTest.java b/geode-core/src/test/java/org/apache/geode/cache/asyncqueue/internal/ParallelAsyncEventQueueImplTest.java
index cf529a6..b85149c 100644
--- a/geode-core/src/test/java/org/apache/geode/cache/asyncqueue/internal/ParallelAsyncEventQueueImplTest.java
+++ b/geode-core/src/test/java/org/apache/geode/cache/asyncqueue/internal/ParallelAsyncEventQueueImplTest.java
@@ -54,8 +54,8 @@ public class ParallelAsyncEventQueueImplTest {
statisticsClock = mock(StatisticsClock.class);
statsFactory = mock(StatisticsFactory.class);
attrs = new GatewaySenderAttributes();
- attrs.isParallel = true;
- attrs.id = "AsyncEventQueue_";
+ attrs.setParallel(true);
+ attrs.setId("AsyncEventQueue_");
InternalDistributedSystem system = mock(InternalDistributedSystem.class);
when(cache.getInternalDistributedSystem()).thenReturn(system);
diff --git a/geode-docs/tools_modules/gfsh/command-pages/alter.html.md.erb b/geode-docs/tools_modules/gfsh/command-pages/alter.html.md.erb
index 98e681c..0b559d8 100644
--- a/geode-docs/tools_modules/gfsh/command-pages/alter.html.md.erb
+++ b/geode-docs/tools_modules/gfsh/command-pages/alter.html.md.erb
@@ -31,6 +31,10 @@ Modify an existing <%=vars.product_name%> resource.
Modifies or removes a region from an offline disk-store.
+- **[alter gateway-sender](#topic_alter_gateway_sender)**
+
+ Modifies attributes of a gateway-sender.
+
- **[alter query-service](#topic_alter_query_service)**
Alter configuration details of the query configuration service.
@@ -183,6 +187,81 @@ alter disk-store --name=DiskStore1 --region=region1 --disk-dirs=/Disks/DiskStore
alter disk-store --name=DiskStore1 --region=region1 --disk-dirs=/Disks/DiskStore1 --remove
```
+## <a id="topic_alter_gateway_sender" class="no-quick-link"></a>alter gateway-sender
+
+Alter attributes of a specified gateway sender on one or more members of a cluster.
+
+**Availability:** Online. You must be connected in <span class="keyword parmname">gfsh</span> to a JMX Manager member to use this command.
+
+**Syntax:**
+
+``` pre
+alter gateway-sender --id=value [--members=value(,value)*] [--groups=value(,value)*]
+ [--alert-threshold=value] [--batch-size=value] [--batch-time-interval=value]
+ [--gateway-event-filter=value(,value)*] [--group-transaction-events(=value)?]
+```
+
+The required option, `--id`, identifies the gateway sender to be altered.
+
+**Parameters, alter gateway-sender**
+<table>
+<colgroup>
+<col width="50%" />
+<col width="50%" />
+</colgroup>
+<thead>
+<tr class="header">
+<th>Name</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td><span class="keyword parmname">\-\-id</span></td>
+<td><em>Required.</em> ID of the gateway sender to be changed</td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-members</span></td>
+<td>Name or ID of the member(s) whose configuration is to be altered at runtime. If you do not specify this parameter, the configuration properties are modified for all cluster members using the cluster configuration service.</td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-groups</span></td>
+<td>Name of the group(s) whose members' runtime configurations are to be altered. If you do not specify this parameter, the configuration properties are modified for all cluster members using the cluster configuration service.</td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-alert-threshold</span></td>
+<td>Maximum time, in milliseconds, that a region event can remain in the gateway sender queue before <%=vars.product_name%> logs an alert.</td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-batch-size</span></td>
+<td>Maximum number of messages that a batch can contain.</td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-batch-time-interval</span></td>
+<td>Maximum time, in milliseconds, that can elapse between sending batches.</td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-gateway-event-filter</span></td>
+<td>A list of fully-qualified class names of GatewayEventFilters (separated by commas) to be associated with the GatewaySender. This serves as a callback for users to filter out events before dispatching to a remote cluster. For example:
+<pre class="pre codeblock"><code>gateway-event-filter=com.user.filters.MyFilter1,com.user.filters.MyFilters2</code></pre>
+<p>In case no value is provided, all existing filters will be removed.</p></td>
+</tr>
+<tr>
+<td><span class="keyword parmname">\-\-group-transaction-events</span></td>
+<td>Boolean value to ensure that all the events of a transaction are sent in the same batch, i.e., they are never spread across different batches.
+<p>Only allowed to be set on gateway senders with the <code class="ph codeph">parallel</code> attribute set to false and <code class="ph codeph">dispatcher-threads</code> attribute equal to 1, or on gateway senders with the <code class="ph codeph">parallel</code> attribute set to true. Also, the <code class="ph codeph">enable-batch-conflation</code> attribute of the gateway sender must be set to false.</p>
+<p><b>Note:</b> In order to work for a transaction, the regions to which the transaction events belong must be replicated by the same set of senders with this flag enabled.</p></td>
+</td>
+</tr>
+</tbody>
+</table>
+
+**Example Commands:**
+
+``` pre
+alter gateway-sender --id=sender1 --alert-threshold=100 --batch-size=200 --group-transaction-events
+```
+
## <a id="topic_alter_query_service" class="no-quick-link"></a>alter query-service
Alter configuration details of the query configuration service.
diff --git a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/AlterGatewaySenderCommand.java b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/AlterGatewaySenderCommand.java
new file mode 100644
index 0000000..16c07dd
--- /dev/null
+++ b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/AlterGatewaySenderCommand.java
@@ -0,0 +1,276 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.Logger;
+import org.springframework.shell.core.annotation.CliCommand;
+import org.springframework.shell.core.annotation.CliOption;
+
+import org.apache.geode.cache.configuration.CacheConfig;
+import org.apache.geode.cache.configuration.DeclarableType;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.InternalConfigurationPersistenceService;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.cli.CliMetaData;
+import org.apache.geode.management.cli.ConverterHint;
+import org.apache.geode.management.cli.SingleGfshCommand;
+import org.apache.geode.management.configuration.ClassName;
+import org.apache.geode.management.internal.cli.functions.AlterGatewaySenderFunction;
+import org.apache.geode.management.internal.cli.functions.GatewaySenderFunctionArgs;
+import org.apache.geode.management.internal.cli.result.model.ResultModel;
+import org.apache.geode.management.internal.exceptions.EntityNotFoundException;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+import org.apache.geode.management.internal.security.ResourceOperation;
+import org.apache.geode.security.ResourcePermission;
+
+public class AlterGatewaySenderCommand extends SingleGfshCommand {
+ private final AlterGatewaySenderFunction alterGatewaySenderFunction =
+ new AlterGatewaySenderFunction();
+ private static final Logger logger = LogService.getLogger();
+
+ @CliCommand(value = CliStrings.ALTER_GATEWAYSENDER,
+ help = CliStrings.ALTER_GATEWAYSENDER__HELP)
+ @CliMetaData(relatedTopic = CliStrings.TOPIC_GEODE_WAN)
+ @ResourceOperation(resource = ResourcePermission.Resource.CLUSTER,
+ operation = ResourcePermission.Operation.MANAGE, target = ResourcePermission.Target.GATEWAY)
+
+ public ResultModel alterGatewaySender(@CliOption(key = CliStrings.ALTER_GATEWAYSENDER__ID,
+ mandatory = true, optionContext = ConverterHint.GATEWAY_SENDER_ID,
+ help = CliStrings.ALTER_GATEWAYSENDER__ID__HELP) String senderId,
+ @CliOption(key = {CliStrings.GROUP, CliStrings.GROUPS},
+ optionContext = ConverterHint.MEMBERGROUP,
+ help = CliStrings.ALTER_GATEWAYSENDER__GROUP__HELP) String[] onGroup,
+ @CliOption(key = {CliStrings.MEMBER, CliStrings.MEMBERS},
+ optionContext = ConverterHint.MEMBERIDNAME,
+ help = CliStrings.ALTER_GATEWAYSENDER__MEMBER__HELP) String[] onMember,
+ @CliOption(key = CliStrings.ALTER_GATEWAYSENDER__ALERTTHRESHOLD,
+ help = CliStrings.ALTER_GATEWAYSENDER__ALERTTHRESHOLD__HELP) Integer alertThreshold,
+ @CliOption(key = CliStrings.ALTER_GATEWAYSENDER__BATCHSIZE,
+ help = CliStrings.ALTER_GATEWAYSENDER__BATCHSIZE__HELP) Integer batchSize,
+ @CliOption(key = CliStrings.ALTER_GATEWAYSENDER__BATCHTIMEINTERVAL,
+ help = CliStrings.ALTER_GATEWAYSENDER__BATCHTIMEINTERVAL__HELP) Integer batchTimeInterval,
+ @CliOption(key = CliStrings.ALTER_GATEWAYSENDER__GATEWAYEVENTFILTER,
+ specifiedDefaultValue = "",
+ // split the input only with comma outside of json string
+ optionContext = "splittingRegex=,(?![^{]*\\})",
+ help = CliStrings.ALTER_GATEWAYSENDER__GATEWAYEVENTFILTER__HELP) ClassName[] gatewayEventFilters,
+ @CliOption(key = CliStrings.ALTER_GATEWAYSENDER__GROUPTRANSACTIONEVENTS,
+ specifiedDefaultValue = "true",
+ help = CliStrings.ALTER_GATEWAYSENDER__GROUPTRANSACTIONEVENTS__HELP) Boolean groupTransactionEvents)
+ throws EntityNotFoundException {
+
+ // need not check if any running servers has this gateway-sender. A server with this
+ // gateway-sender id
+ // may be shutdown, but we still need to update Cluster Configuration.
+ if (getConfigurationPersistenceService() == null) {
+ return ResultModel.createError("Cluster Configuration Service is not available. "
+ + "Please connect to a locator with running Cluster Configuration Service.");
+ }
+
+ final String id = senderId.trim();
+
+ CacheConfig.GatewaySender oldConfiguration = findGatewaySenderConfiguration(id);
+
+ if (oldConfiguration == null) {
+ String message = String.format("Cannot find a gateway sender with id '%s'.", id);
+ throw new EntityNotFoundException(message);
+ }
+
+ if (alertThreshold != null && alertThreshold < 0) {
+ return ResultModel.createError(
+ "alter-gateway-sender cannot be performed for --alert-threshold values smaller then 0.");
+ }
+
+ if (batchSize != null && batchSize < 1) {
+ return ResultModel.createError(
+ "alter-gateway-sender cannot be performed for --batch-size values smaller then 1.");
+ }
+
+ if (batchTimeInterval != null && batchTimeInterval < -1) {
+ return ResultModel.createError(
+ "alter-gateway-sender cannot be performed for --batch-time-interval values smaller then -1.");
+ }
+
+ if (groupTransactionEvents != null && groupTransactionEvents
+ && !oldConfiguration.mustGroupTransactionEvents()) {
+ if (!oldConfiguration.isParallel() && (oldConfiguration.getDispatcherThreads() == null
+ || Integer.parseInt(oldConfiguration.getDispatcherThreads()) > 1)) {
+ return ResultModel.createError(
+ "alter-gateway-sender cannot be performed for --group-transaction-events attribute if serial sender and dispatcher-threads is greater than 1.");
+ }
+
+ if (oldConfiguration.isEnableBatchConflation()) {
+ return ResultModel.createError(
+ "alter-gateway-sender cannot be performed for --group-transaction-events attribute if batch-conflation is enabled.");
+ }
+ }
+
+ Set<DistributedMember> dsMembers = findMembers(onGroup, onMember);
+
+ if (dsMembers.isEmpty()) {
+ return ResultModel.createError(CliStrings.NO_MEMBERS_FOUND_MESSAGE);
+ }
+
+ // Don't allow alter gateway sender command to be performed if all members are not the current
+ // version.
+ if (!verifyAllCurrentVersion(dsMembers)) {
+ return ResultModel.createError(
+ CliStrings.ALTER_GATEWAYSENDER__MSG__CAN_NOT_CREATE_DIFFERENT_VERSIONS);
+ }
+
+ CacheConfig.GatewaySender gwConfiguration = new CacheConfig.GatewaySender();
+ gwConfiguration.setId(id);
+
+ boolean modify = false;
+
+ if (alertThreshold != null) {
+ modify = true;
+ gwConfiguration.setAlertThreshold(alertThreshold.toString());
+ }
+
+ if (batchSize != null) {
+ modify = true;
+ gwConfiguration.setBatchSize(batchSize.toString());
+ }
+
+ if (batchTimeInterval != null) {
+ modify = true;
+ gwConfiguration.setBatchTimeInterval(batchTimeInterval.toString());
+ }
+
+ if (groupTransactionEvents != null) {
+ modify = true;
+ gwConfiguration.setGroupTransactionEvents(groupTransactionEvents);
+ }
+
+
+ if (gatewayEventFilters != null) {
+ modify = true;
+ if (gatewayEventFilters.length == 1
+ && gatewayEventFilters[0].getClassName().isEmpty()) {
+ gwConfiguration.getGatewayEventFilters();
+ } else {
+ gwConfiguration.getGatewayEventFilters()
+ .addAll(Arrays.stream(gatewayEventFilters)
+ .map(l -> new DeclarableType(l.getClassName()))
+ .collect(Collectors.toList()));
+ }
+ }
+
+ if (!modify) {
+ return ResultModel.createError(CliStrings.ALTER_GATEWAYSENDER__RELEVANT__OPTION__MESSAGE);
+ }
+
+ GatewaySenderFunctionArgs gatewaySenderFunctionArgs =
+ new GatewaySenderFunctionArgs(gwConfiguration);
+
+ List<CliFunctionResult> gatewaySenderAlterResults =
+ executeAndGetFunctionResult(alterGatewaySenderFunction, gatewaySenderFunctionArgs,
+ dsMembers);
+
+ ResultModel resultModel = ResultModel.createMemberStatusResult(gatewaySenderAlterResults);
+
+ resultModel.setConfigObject(gwConfiguration);
+
+ return resultModel;
+ }
+
+ @Override
+ public boolean updateConfigForGroup(String group, CacheConfig config, Object configObject) {
+ List<CacheConfig.GatewaySender> gwSenders = config.getGatewaySenders();
+ if (gwSenders.isEmpty()) {
+ return false;
+ }
+
+ boolean gwConfigsHaveBeenUpdated = false;
+ CacheConfig.GatewaySender gwConfiguration =
+ ((CacheConfig.GatewaySender) configObject);
+
+ String gwId = gwConfiguration.getId();
+
+ for (CacheConfig.GatewaySender sender : gwSenders) {
+ if (gwId.equals(sender.getId())) {
+ gwConfigsHaveBeenUpdated = true;
+ if (StringUtils.isNotBlank(gwConfiguration.getBatchSize())) {
+ sender.setBatchSize(gwConfiguration.getBatchSize());
+ }
+
+ if (StringUtils.isNotBlank(gwConfiguration.getBatchTimeInterval())) {
+ sender.setBatchTimeInterval(gwConfiguration.getBatchTimeInterval());
+ }
+
+ if (StringUtils.isNotBlank(gwConfiguration.getAlertThreshold())) {
+ sender.setAlertThreshold(gwConfiguration.getAlertThreshold());
+ }
+ if (gwConfiguration.mustGroupTransactionEvents() != null) {
+ sender.setGroupTransactionEvents(gwConfiguration.mustGroupTransactionEvents());
+ }
+
+ if (gwConfiguration.areGatewayEventFiltersUpdated()) {
+ if (!sender.getGatewayEventFilters().isEmpty()) {
+ sender.getGatewayEventFilters().clear();
+ }
+ if (!gwConfiguration.getGatewayEventFilters().isEmpty()) {
+ sender.getGatewayEventFilters().addAll(gwConfiguration.getGatewayEventFilters());
+ }
+ }
+
+ }
+ }
+ return gwConfigsHaveBeenUpdated;
+
+ }
+
+ private CacheConfig.GatewaySender findGatewaySenderConfiguration(String gwId) {
+ CacheConfig.GatewaySender gwsender = null;
+ InternalConfigurationPersistenceService ccService = this.getConfigurationPersistenceService();
+ if (ccService == null) {
+ return null;
+ }
+
+ Set<String> groups = ccService.getGroups();
+
+ for (String group : groups) {
+ List<CacheConfig.GatewaySender> gwSendersList =
+ ccService.getCacheConfig(group).getGatewaySenders();
+ if (gwSendersList.isEmpty())
+ continue;
+
+ for (CacheConfig.GatewaySender sender : gwSendersList) {
+ if (sender.getId().equals(gwId))
+ return sender;
+ }
+
+ }
+ return gwsender;
+ }
+
+ boolean verifyAllCurrentVersion(Set<DistributedMember> members) {
+ return members.stream().allMatch(
+ member -> ((InternalDistributedMember) member).getVersion()
+ .equals(KnownVersion.CURRENT));
+ }
+
+}
diff --git a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/CommandAvailabilityIndicator.java b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/CommandAvailabilityIndicator.java
index ae0b7c4..37d1268 100644
--- a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/CommandAvailabilityIndicator.java
+++ b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/commands/CommandAvailabilityIndicator.java
@@ -45,7 +45,7 @@ public class CommandAvailabilityIndicator extends GfshCommand {
CliStrings.STATUS_SHARED_CONFIG, CliStrings.CREATE_GATEWAYSENDER,
CliStrings.START_GATEWAYSENDER, CliStrings.PAUSE_GATEWAYSENDER,
CliStrings.RESUME_GATEWAYSENDER, CliStrings.RESUME_ASYNCEVENTQUEUE,
- CliStrings.STOP_GATEWAYSENDER,
+ CliStrings.STOP_GATEWAYSENDER, CliStrings.ALTER_GATEWAYSENDER,
CliStrings.CREATE_GATEWAYRECEIVER, CliStrings.START_GATEWAYRECEIVER,
CliStrings.STOP_GATEWAYRECEIVER, CliStrings.LIST_GATEWAY, CliStrings.STATUS_GATEWAYSENDER,
CliStrings.STATUS_GATEWAYRECEIVER, CliStrings.LOAD_BALANCE_GATEWAYSENDER,
diff --git a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunction.java b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunction.java
new file mode 100644
index 0000000..b9cb597
--- /dev/null
+++ b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunction.java
@@ -0,0 +1,122 @@
+/*
+ * 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.util.ArrayList;
+import java.util.List;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.execute.ResultSender;
+import org.apache.geode.cache.wan.GatewayEventFilter;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.execute.InternalFunction;
+import org.apache.geode.internal.security.CallbackInstantiator;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.management.internal.exceptions.EntityNotFoundException;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.management.internal.i18n.CliStrings;
+
+public class AlterGatewaySenderFunction implements InternalFunction<GatewaySenderFunctionArgs> {
+ private static final long serialVersionUID = 1L;
+
+ private static final String ID = AlterGatewaySenderFunction.class.getName();
+
+ private static final Logger logger = LogService.getLogger();
+
+ @Override
+ public void execute(FunctionContext<GatewaySenderFunctionArgs> context) {
+ ResultSender<Object> resultSender = context.getResultSender();
+
+ Cache cache = context.getCache();
+ String memberNameOrId = context.getMemberName();
+
+ GatewaySenderFunctionArgs gatewaySenderAlterArgs =
+ context.getArguments();
+
+ try {
+ GatewaySender alterGatewaySender = alterGatewaySender(cache, gatewaySenderAlterArgs);
+ resultSender.lastResult(new CliFunctionResult(memberNameOrId,
+ CliFunctionResult.StatusState.OK, CliStrings.format(
+ CliStrings.GATEWAY_SENDER_0_IS_UPDATED_ON_MEMBER_1,
+ alterGatewaySender.getId(), memberNameOrId)));
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ resultSender.lastResult(new CliFunctionResult(memberNameOrId, e, null));
+ }
+ }
+
+ private GatewaySender alterGatewaySender(Cache cache,
+ GatewaySenderFunctionArgs gatewaySenderCreateArgs) {
+ String gwId = gatewaySenderCreateArgs.getId();
+ GatewaySender gateway = cache.getGatewaySender(gwId);
+ if (gateway == null) {
+ String message = String.format("Cannot find existing gateway sender with id '%s'.", gwId);
+ throw new EntityNotFoundException(message);
+ }
+
+ boolean pause = false;
+ if (gateway.isRunning() && !gateway.isPaused()) {
+ gateway.pause();
+ pause = true;
+ }
+
+ Integer alertThreshold = gatewaySenderCreateArgs.getAlertThreshold();
+ if (alertThreshold != null) {
+ gateway.setAlertThreshold(alertThreshold.intValue());
+ }
+
+ Integer batchSize = gatewaySenderCreateArgs.getBatchSize();
+ if (batchSize != null) {
+ gateway.setBatchSize(batchSize.intValue());
+ }
+
+ Integer batchTimeInterval = gatewaySenderCreateArgs.getBatchTimeInterval();
+ if (batchTimeInterval != null) {
+ gateway.setBatchTimeInterval(batchTimeInterval.intValue());
+ }
+
+ Boolean groupTransactionEvents = gatewaySenderCreateArgs.mustGroupTransactionEvents();
+ if (groupTransactionEvents != null) {
+ gateway.setGroupTransactionEvents(groupTransactionEvents.booleanValue());
+ }
+
+ List<String> gatewayEventFilters = gatewaySenderCreateArgs.getGatewayEventFilter();
+ if (gatewayEventFilters != null) {
+ List<GatewayEventFilter> filters = new ArrayList<>();
+ if (!gatewayEventFilters.isEmpty()) {
+ for (String filter : gatewayEventFilters) {
+ filters.add(CallbackInstantiator.getObjectOfTypeFromClassName(filter,
+ GatewayEventFilter.class));
+ }
+ }
+ gateway.setGatewayEventFilters(filters);
+ }
+
+ if (pause) {
+ gateway.resume();
+ }
+ return gateway;
+ }
+
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+}
diff --git a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/GatewaySenderFunctionArgs.java b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/GatewaySenderFunctionArgs.java
index dd70bda..af2e316 100644
--- a/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/GatewaySenderFunctionArgs.java
+++ b/geode-gfsh/src/main/java/org/apache/geode/management/internal/cli/functions/GatewaySenderFunctionArgs.java
@@ -15,6 +15,7 @@
package org.apache.geode.management.internal.cli.functions;
import java.io.Serializable;
+import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -66,12 +67,17 @@ public class GatewaySenderFunctionArgs implements Serializable {
this.alertThreshold = string2int(sender.getAlertThreshold());
this.dispatcherThreads = string2int(sender.getDispatcherThreads());
this.orderPolicy = sender.getOrderPolicy();
- this.gatewayEventFilters =
- Optional.of(sender.getGatewayEventFilters())
- .map(filters -> filters
- .stream().map(DeclarableType::getClassName)
- .collect(Collectors.toList()))
- .orElse(null);
+ if (sender.areGatewayEventFiltersUpdated()) {
+ this.gatewayEventFilters =
+ Optional.of(sender.getGatewayEventFilters())
+ .map(filters -> filters
+ .stream().map(DeclarableType::getClassName)
+ .collect(Collectors.toList()))
+ .orElse(Collections.emptyList());
+ } else {
+ this.gatewayEventFilters = null;
+ }
+
this.gatewayTransportFilters =
Optional.of(sender.getGatewayTransportFilters())
.map(filters -> filters
diff --git a/geode-gfsh/src/main/resources/org/apache/geode/gfsh/internal/management/sanctioned-geode-gfsh-serializables.txt b/geode-gfsh/src/main/resources/org/apache/geode/gfsh/internal/management/sanctioned-geode-gfsh-serializables.txt
index 8cd7937..656248a 100644
--- a/geode-gfsh/src/main/resources/org/apache/geode/gfsh/internal/management/sanctioned-geode-gfsh-serializables.txt
+++ b/geode-gfsh/src/main/resources/org/apache/geode/gfsh/internal/management/sanctioned-geode-gfsh-serializables.txt
@@ -30,6 +30,7 @@ org/apache/geode/management/internal/cli/domain/RegionDescription,true,646144927
org/apache/geode/management/internal/cli/domain/RegionDescriptionPerMember,true,1,hostingMember:java/lang/String,isAccessor:boolean,name:java/lang/String,regionAttributesInfo:org/apache/geode/management/internal/cli/domain/RegionAttributesInfo,size:int
org/apache/geode/management/internal/cli/domain/RegionInformation,true,1,dataPolicy:org/apache/geode/cache/DataPolicy,isRoot:boolean,name:java/lang/String,parentRegion:java/lang/String,path:java/lang/String,scope:org/apache/geode/cache/Scope,subRegionInformationSet:java/util/Set
org/apache/geode/management/internal/cli/domain/StackTracesPerMember,true,1,memberNameOrId:java/lang/String,stackTraces:byte[],timestamp:java/time/Instant
+org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunction,true,1
org/apache/geode/management/internal/cli/functions/AlterQueryServiceFunction,true,7155576168386556341
org/apache/geode/management/internal/cli/functions/AlterRuntimeConfigFunction,true,1
org/apache/geode/management/internal/cli/functions/AsyncEventQueueFunctionArgs,true,-6524494645663740872,asyncEventQueueId:java/lang/String,batchSize:int,batchTimeInterval:int,diskStoreName:java/lang/String,diskSynchronous:boolean,dispatcherThreads:int,enableBatchConflation:boolean,forwardExpirationDestroy:boolean,gatewayEventFilters:java/lang/String[],gatewaySubstitutionFilter:java/lang/String,isParallel:boolean,listenerClassName:java/lang/String,listenerProperties:java/util/Properties, [...]
diff --git a/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/AlterGatewaySenderCommandTest.java b/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/AlterGatewaySenderCommandTest.java
new file mode 100644
index 0000000..019e529
--- /dev/null
+++ b/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/AlterGatewaySenderCommandTest.java
@@ -0,0 +1,138 @@
+/*
+ * 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.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.geode.cache.configuration.CacheConfig;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.InternalConfigurationPersistenceService;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.config.JAXBService;
+import org.apache.geode.management.internal.cli.functions.GatewaySenderFunctionArgs;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+import org.apache.geode.test.junit.rules.GfshParserRule;
+
+public class AlterGatewaySenderCommandTest {
+ @ClassRule
+ public static GfshParserRule gfsh = new GfshParserRule();
+
+ private AlterGatewaySenderCommand command;
+ private InternalConfigurationPersistenceService service;
+ private Set<String> groupSet = new HashSet<>();
+
+ private List<CliFunctionResult> functionResults;
+ private CliFunctionResult cliFunctionResult;
+ private ArgumentCaptor<GatewaySenderFunctionArgs> argsArgumentCaptor =
+ ArgumentCaptor.forClass(GatewaySenderFunctionArgs.class);
+
+
+ @Before
+ public void before() {
+ command = spy(AlterGatewaySenderCommand.class);
+ InternalCache cache = mock(InternalCache.class);
+ doReturn(cache).when(command).getCache();
+ service =
+ spy(new InternalConfigurationPersistenceService(JAXBService.create(CacheConfig.class)));
+
+ doReturn(service).when(command).getConfigurationPersistenceService();
+
+ groupSet.add("group1");
+ groupSet.add("group2");
+ doReturn(groupSet).when(service).getGroups();
+
+ CacheConfig config = new CacheConfig();
+ CacheConfig.GatewaySender gw1 = new CacheConfig.GatewaySender();
+ gw1.setId("sender1");
+ gw1.setParallel(true);
+ gw1.setGroupTransactionEvents(false);
+ gw1.setEnableBatchConflation(true);
+
+ CacheConfig.GatewaySender gw2 = new CacheConfig.GatewaySender();
+ gw2.setId("sender2");
+ gw2.setParallel(false);
+ gw2.setDispatcherThreads("5");
+ gw2.setGroupTransactionEvents(false);
+ gw2.setEnableBatchConflation(false);
+
+ config.getGatewaySenders().add(gw1);
+ config.getGatewaySenders().add(gw2);
+
+ doReturn(config).when(service).getCacheConfig("group1");
+ doReturn(new CacheConfig()).when(service).getCacheConfig("group2");
+
+ Set<DistributedMember> members =
+ Stream.of(mock(DistributedMember.class)).collect(Collectors.toSet());
+ doReturn(members).when(command).findMembers(any(), any());
+
+ doReturn(true).when(command).verifyAllCurrentVersion(members);
+
+ }
+
+ @Test
+ public void mandatoryOption() {
+ gfsh.executeAndAssertThat(command, "alter gateway-sender").statusIsError()
+ .containsOutput("Invalid command");
+ }
+
+ @Test
+ public void noOptionToModify() {
+ gfsh.executeAndAssertThat(command, "alter gateway-sender --id=sender1").statusIsError()
+ .containsOutput("Please provide a relevant parameter(s)");
+ }
+
+ @Test
+ public void unknownOption() {
+ gfsh.executeAndAssertThat(command, "alter gateway-sender --id=sender1 --status").statusIsError()
+ .containsOutput("Invalid command");
+ }
+
+ @Test
+ public void emptyConfiguration() {
+ gfsh.executeAndAssertThat(command, "alter gateway-sender --id=test --batch-size=100")
+ .statusIsError().containsOutput("Cannot find a gateway sender");
+ }
+
+ @Test
+ public void changeGroupTransaction1() {
+ gfsh.executeAndAssertThat(command,
+ "alter gateway-sender --id=sender1 --group-transaction-events").statusIsError()
+ .containsOutput("alter-gateway-sender cannot be performed");
+ }
+
+ @Test
+ public void changeGroupTransaction2() {
+ gfsh.executeAndAssertThat(command,
+ "alter gateway-sender --id=sender2 --group-transaction-events").statusIsError()
+ .containsOutput("alter-gateway-sender cannot be performed");
+ }
+
+}
diff --git a/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/CreateGatewaySenderCommandTest.java b/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/CreateGatewaySenderCommandTest.java
index 585158d..e7b56d3 100644
--- a/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/CreateGatewaySenderCommandTest.java
+++ b/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/commands/CreateGatewaySenderCommandTest.java
@@ -272,7 +272,7 @@ public class CreateGatewaySenderCommandTest {
assertThat(argsArgumentCaptor.getValue().getAlertThreshold()).isNull();
assertThat(argsArgumentCaptor.getValue().getDispatcherThreads()).isNull();
assertThat(argsArgumentCaptor.getValue().getOrderPolicy()).isNull();
- assertThat(argsArgumentCaptor.getValue().getGatewayEventFilter()).isNotNull().isEmpty();
+ assertThat(argsArgumentCaptor.getValue().getGatewayEventFilter()).isNull();
assertThat(argsArgumentCaptor.getValue().getGatewayTransportFilter()).isNotNull().isEmpty();
assertThat(argsArgumentCaptor.getValue().mustGroupTransactionEvents()).isNotNull();
assertThat(argsArgumentCaptor.getValue().getEnforceThreadsConnectSameReceiver()).isFalse();
diff --git a/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunctionTest.java b/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunctionTest.java
new file mode 100644
index 0000000..8d8ae54
--- /dev/null
+++ b/geode-gfsh/src/test/java/org/apache/geode/management/internal/cli/functions/AlterGatewaySenderFunctionTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.execute.ResultSender;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.distributed.DistributedSystem;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.wan.GatewaySenderAttributes;
+import org.apache.geode.management.internal.exceptions.EntityNotFoundException;
+import org.apache.geode.management.internal.functions.CliFunctionResult;
+
+
+public class AlterGatewaySenderFunctionTest {
+
+ private AlterGatewaySenderFunction function;
+ private FunctionContext<GatewaySenderFunctionArgs> context;
+ private InternalCache cache;
+ private ResultSender<Object> resultSender;
+ private ArgumentCaptor<CliFunctionResult> resultCaptor;
+ private GatewaySenderFunctionArgs args;
+ private GatewaySender sender;
+ private ArgumentCaptor<GatewaySenderAttributes> updateCapture1;
+
+ @Before
+ @SuppressWarnings("unchecked")
+ public void before() {
+ sender = mock(GatewaySender.class);
+ function = spy(AlterGatewaySenderFunction.class);
+ context = mock(FunctionContext.class);
+ cache = mock(InternalCache.class);
+ args = mock(GatewaySenderFunctionArgs.class);
+ resultSender = mock(ResultSender.class);
+ when(context.getCache()).thenReturn(cache);
+ when(context.getResultSender()).thenReturn(resultSender);
+ when(context.getArguments()).thenReturn(args);
+ when(context.getMemberName()).thenReturn("member");
+ when(args.getId()).thenReturn("id");
+ resultCaptor = ArgumentCaptor.forClass(CliFunctionResult.class);
+ when(cache.getDistributedSystem()).thenReturn(mock(DistributedSystem.class));
+ when(sender.getId()).thenReturn("id");
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void gateWaySenderNotFound() {
+ when(cache.getGatewaySender(any())).thenReturn(null);
+ function.execute(context);
+
+ verify(resultSender).lastResult(resultCaptor.capture());
+ CliFunctionResult result = resultCaptor.getValue();
+ assertThat(result.isSuccessful()).isFalse();
+ assertThat(result.getThrowable()).isInstanceOf(EntityNotFoundException.class);
+ assertThat(result.getMessage()).isNull();
+ }
+
+ @Test
+ @SuppressWarnings({"unchecked", "deprecation"})
+ public void gateWaySenderNoAttributesToUpdate() {
+ when(cache.getGatewaySender("id")).thenReturn(sender);
+ when(args.getAlertThreshold()).thenReturn(null);
+ when(args.getBatchSize()).thenReturn(null);
+ when(args.getBatchTimeInterval()).thenReturn(null);
+ when(args.getGatewayEventFilter()).thenReturn(null);
+ when(args.mustGroupTransactionEvents()).thenReturn(null);
+
+ function.execute(context);
+
+ verify(resultSender).lastResult(resultCaptor.capture());
+ CliFunctionResult result = resultCaptor.getValue();
+ assertThat(result.isSuccessful()).isTrue();
+ assertThat(result.getThrowable()).isNull();
+ assertThat(result.getMessage()).isEqualTo("GatewaySender id is updated on member member");
+ }
+
+ @Test
+ @SuppressWarnings({"unchecked", "deprecation"})
+ public void gateWaySender3AttributesUpdated() {
+ when(cache.getGatewaySender("id")).thenReturn(sender);
+ when(args.getAlertThreshold()).thenReturn(200);
+ when(args.getBatchSize()).thenReturn(50);
+ when(args.getBatchTimeInterval()).thenReturn(150);
+ when(args.getGatewayEventFilter()).thenReturn(null);
+ when(args.mustGroupTransactionEvents()).thenReturn(null);
+
+ function.execute(context);
+ verify(sender).setAlertThreshold(200);
+ verify(sender).setBatchSize(50);
+ verify(sender).setBatchTimeInterval(150);
+
+ verify(resultSender).lastResult(resultCaptor.capture());
+ CliFunctionResult result = resultCaptor.getValue();
+ assertThat(result.isSuccessful()).isTrue();
+ assertThat(result.getThrowable()).isNull();
+ assertThat(result.getMessage()).isEqualTo("GatewaySender id is updated on member member");
+ }
+}
diff --git a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
index 6b76e6a..b328af9 100644
--- a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
+++ b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
@@ -53,6 +53,7 @@ import org.apache.geode.GemFireIOException;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.wan.GatewayEventFilter;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.InternalCache;
@@ -94,6 +95,9 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase {
@Before
public void setUp() throws Exception {
addIgnoredException("Broken pipe||Unexpected IOException");
+ addIgnoredException("Connection reset");
+ addIgnoredException("Connection refused");
+ addIgnoredException("could not get remote locator information");
}
@Test(timeout = 300_000)
@@ -971,6 +975,217 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase {
});
}
+ @Test
+ public void testParallelGWSenderUpdateAttrWhileEntriesInQueue() throws Exception {
+ Integer[] locatorPorts = createLNAndNYLocators();
+ Integer lnPort = locatorPorts[0];
+ Integer nyPort = locatorPorts[1];
+
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+
+ createPartitionedRegions(false);
+
+ vm4.invoke(() -> waitForSenderRunningState("ln"));
+ vm5.invoke(() -> waitForSenderRunningState("ln"));
+ vm6.invoke(() -> waitForSenderRunningState("ln"));
+ vm7.invoke(() -> waitForSenderRunningState("ln"));
+
+ vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000));
+
+ updateBatchSize(50);
+ updateBatchTimeInterval(100);
+
+ createReceiverInVMs(vm2, vm3);
+
+ validateRegionSizes(getUniqueName() + "_PR", 1000, vm2, vm3);
+
+ checkBatchSize(50);
+ checkBatchTimeInterval(100);
+ }
+
+ @Test
+ public void testParallelGWSenderUpdateAttrWhilePutting() throws Exception {
+ Integer[] locatorPorts = createLNAndNYLocators();
+ Integer lnPort = locatorPorts[0];
+ Integer nyPort = locatorPorts[1];
+
+ createCacheInVMs(nyPort, vm2, vm3);
+ createReceiverInVMs(vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+
+ createPartitionedRegions(false);
+
+ vm4.invoke(() -> waitForSenderRunningState("ln"));
+ vm5.invoke(() -> waitForSenderRunningState("ln"));
+ vm6.invoke(() -> waitForSenderRunningState("ln"));
+ vm7.invoke(() -> waitForSenderRunningState("ln"));
+
+ AsyncInvocation async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000));
+
+ updateBatchSize(100);
+ updateBatchTimeInterval(150);
+
+ async.await();
+ validateRegionSizes(getUniqueName() + "_PR", 5000, vm2, vm3);
+
+ checkBatchSize(100);
+ checkBatchTimeInterval(150);
+ }
+
+ @Test
+ public void testParallelGWSenderUpdateAttrWhilePaused() throws Exception {
+ Integer[] locatorPorts = createLNAndNYLocators();
+ Integer lnPort = locatorPorts[0];
+ Integer nyPort = locatorPorts[1];
+
+ createCacheInVMs(nyPort, vm2, vm3);
+ createReceiverInVMs(vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+
+ createPartitionedRegions(false);
+
+ vm4.invoke(() -> waitForSenderRunningState("ln"));
+ vm5.invoke(() -> waitForSenderRunningState("ln"));
+ vm6.invoke(() -> waitForSenderRunningState("ln"));
+ vm7.invoke(() -> waitForSenderRunningState("ln"));
+
+ vm4.invoke(() -> pauseSender("ln"));
+ vm5.invoke(() -> pauseSender("ln"));
+ vm6.invoke(() -> pauseSender("ln"));
+ vm7.invoke(() -> pauseSender("ln"));
+
+ vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 5000));
+
+ updateGroupTransactionEvents(true);
+ updateBatchTimeInterval(200);
+
+ validateRegionSizes(getUniqueName() + "_PR", 0, vm2, vm3);
+
+ checkGroupTransactionEvents(true);
+ checkBatchTimeInterval(200);
+
+ vm4.invoke(() -> resumeSender("ln"));
+ vm5.invoke(() -> resumeSender("ln"));
+ vm6.invoke(() -> resumeSender("ln"));
+ vm7.invoke(() -> resumeSender("ln"));
+
+ validateRegionSizes(getUniqueName() + "_PR", 5000, vm2, vm3);
+
+ }
+
+ @Test
+ public void testParallelGWSenderUpdateFiltersWhilePutting() throws Exception {
+ Integer[] locatorPorts = createLNAndNYLocators();
+ Integer lnPort = locatorPorts[0];
+ Integer nyPort = locatorPorts[1];
+
+ List<GatewayEventFilter> filters = new ArrayList<>();
+ filters.add(new MyGatewayEventFilter_AfterAck());
+ filters.add(new PDXGatewayEventFilter());
+
+ List<GatewayEventFilter> filters2 = new ArrayList<>();
+ filters.add(new MyGatewayEventFilter());
+ filters.add(new MyGatewayEventFilter_AfterAck());
+ filters.add(new PDXGatewayEventFilter());
+
+ createCacheInVMs(nyPort, vm2, vm3);
+ createReceiverInVMs(vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+
+ createPartitionedRegions(false);
+
+ vm4.invoke(() -> waitForSenderRunningState("ln"));
+ vm5.invoke(() -> waitForSenderRunningState("ln"));
+ vm6.invoke(() -> waitForSenderRunningState("ln"));
+ vm7.invoke(() -> waitForSenderRunningState("ln"));
+
+ AsyncInvocation async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000));
+ checkBatchSize(10);
+
+ updateBatchSize(100);
+ updateGatewayEventFilters(filters);
+ checkBatchSize(100);
+ updateGatewayEventFilters(filters2);
+
+ async.await();
+
+ }
+
+
+ @Test
+ public void testParallelGWSenderUpdateFiltersWhilePuttingOnOneDispatcThread() throws Exception {
+ Integer[] locatorPorts = createLNAndNYLocators();
+ Integer lnPort = locatorPorts[0];
+ Integer nyPort = locatorPorts[1];
+
+ List<GatewayEventFilter> filters = new ArrayList<>();
+ filters.add(new MyGatewayEventFilter_AfterAck());
+ filters.add(new PDXGatewayEventFilter());
+
+ List<GatewayEventFilter> filters2 = new ArrayList<>();
+ filters.add(new MyGatewayEventFilter());
+ filters.add(new MyGatewayEventFilter_AfterAck());
+ filters.add(new PDXGatewayEventFilter());
+
+ createCacheInVMs(nyPort, vm2, vm3);
+ createReceiverInVMs(vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+ vm4.invoke(() -> setNumDispatcherThreadsForTheRun(1));
+ vm5.invoke(() -> setNumDispatcherThreadsForTheRun(1));
+ vm6.invoke(() -> setNumDispatcherThreadsForTheRun(1));
+ vm7.invoke(() -> setNumDispatcherThreadsForTheRun(1));
+
+ vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+ vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false));
+
+ createPartitionedRegions(false);
+
+ vm4.invoke(() -> waitForSenderRunningState("ln"));
+ vm5.invoke(() -> waitForSenderRunningState("ln"));
+ vm6.invoke(() -> waitForSenderRunningState("ln"));
+ vm7.invoke(() -> waitForSenderRunningState("ln"));
+
+ AsyncInvocation async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000));
+ checkBatchSize(10);
+
+ updateBatchSize(100);
+ updateGatewayEventFilters(filters);
+ checkBatchSize(100);
+ updateGatewayEventFilters(filters2);
+
+ async.await();
+
+ }
+
+
private void clearShadowBucketRegions(PartitionedRegion shadowRegion) {
PartitionedRegionDataStore.BucketVisitor bucketVisitor =
new PartitionedRegionDataStore.BucketVisitor() {
@@ -1023,6 +1238,268 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase {
vm3.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap()));
}
+ private void updateBatchSize(int batchsize) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchSize(batchsize);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchSize(batchsize);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchSize(batchsize);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchSize(batchsize);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void updateBatchTimeInterval(int batchTimeInterval) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchTimeInterval(batchTimeInterval);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchTimeInterval(batchTimeInterval);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchTimeInterval(batchTimeInterval);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchTimeInterval(batchTimeInterval);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void updateGroupTransactionEvents(boolean groupTransactionEvents) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGroupTransactionEvents(groupTransactionEvents);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGroupTransactionEvents(groupTransactionEvents);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGroupTransactionEvents(groupTransactionEvents);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGroupTransactionEvents(groupTransactionEvents);
+ if (paused) {
+ sender.resume();
+ }
+ });
+
+ }
+
+ private void updateGatewayEventFilters(List<GatewayEventFilter> filters) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGatewayEventFilters(filters);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGatewayEventFilters(filters);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGatewayEventFilters(filters);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGatewayEventFilters(filters);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void checkBatchSize(int batchsize) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchSize()).isEqualTo(batchsize);
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchSize()).isEqualTo(batchsize);
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchSize()).isEqualTo(batchsize);
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchSize()).isEqualTo(batchsize);
+ });
+ }
+
+ private void checkBatchTimeInterval(int batchTimeInterval) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval);
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval);
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval);
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval);
+ });
+ }
+
+ private void checkGroupTransactionEvents(boolean groupTransactionEvents) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents);
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents);
+ });
+ vm6.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents);
+ });
+ vm7.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents);
+ });
+ }
+
private void stopSenders() {
vm4.invoke(() -> stopSender("ln"));
vm5.invoke(() -> stopSender("ln"));
diff --git a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderAlterOperationsDUnitTest.java b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderAlterOperationsDUnitTest.java
new file mode 100644
index 0000000..929e7a7
--- /dev/null
+++ b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderAlterOperationsDUnitTest.java
@@ -0,0 +1,885 @@
+/*
+ * 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.internal.cache.wan.serial;
+
+import static java.lang.System.currentTimeMillis;
+import static java.util.Arrays.asList;
+import static org.apache.geode.cache.Region.SEPARATOR;
+import static org.apache.geode.cache.RegionShortcut.REPLICATE;
+import static org.apache.geode.cache.wan.GatewaySender.DEFAULT_ORDER_POLICY;
+import static org.apache.geode.cache.wan.GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE;
+import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
+import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
+import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
+import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
+import static org.apache.geode.test.dunit.VM.getCurrentVMNum;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.apache.geode.test.dunit.VM.toArray;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.StringTokenizer;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+import org.apache.logging.log4j.Logger;
+import org.junit.Before;
+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.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
+import org.junit.runners.Parameterized.UseParametersRunnerFactory;
+
+import org.apache.geode.cache.CacheClosedException;
+import org.apache.geode.cache.DataPolicy;
+import org.apache.geode.cache.DiskStore;
+import org.apache.geode.cache.DiskStoreFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionDestroyedException;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.Scope;
+import org.apache.geode.cache.wan.GatewayEventFilter;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewayReceiver;
+import org.apache.geode.cache.wan.GatewayReceiverFactory;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
+import org.apache.geode.distributed.Locator;
+import org.apache.geode.distributed.internal.InternalLocator;
+import org.apache.geode.internal.cache.ForceReattemptException;
+import org.apache.geode.internal.cache.RegionQueue;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
+import org.apache.geode.internal.cache.wan.GatewaySenderException;
+import org.apache.geode.internal.cache.wan.InternalGatewaySender;
+import org.apache.geode.internal.cache.wan.InternalGatewaySenderFactory;
+import org.apache.geode.internal.cache.wan.WANTestBase;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+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.dunit.cache.CacheTestCase;
+import org.apache.geode.test.junit.categories.WanTest;
+import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
+import org.apache.geode.test.junit.runners.CategoryWithParameterizedRunnerFactory;
+
+@Category(WanTest.class)
+@RunWith(Parameterized.class)
+@UseParametersRunnerFactory(CategoryWithParameterizedRunnerFactory.class)
+@SuppressWarnings("serial")
+public class SerialGatewaySenderAlterOperationsDUnitTest extends CacheTestCase {
+ private static final Logger logger = LogService.getLogger();
+
+ @Parameters(name = "{index}: numDispatchers={0}")
+ public static Collection<Integer> data() {
+ return asList(1, 3, 5);
+ }
+
+ @Parameter
+ public int numDispatchers;
+
+ private VM vm0;
+ private VM vm1;
+ private VM vm2;
+ private VM vm3;
+ private VM vm4;
+ private VM vm5;
+ private VM vm6;
+ private VM vm7;
+
+ private String className;
+
+ @Rule
+ public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();
+
+ @Before
+ public void setUp() {
+ addIgnoredException("Broken pipe");
+ addIgnoredException("Connection refused");
+ addIgnoredException("Connection reset");
+ addIgnoredException("could not get remote locator information");
+ addIgnoredException("Software caused connection abort");
+ addIgnoredException("Unexpected IOException");
+
+ className = getClass().getSimpleName();
+
+ vm0 = getVM(0);
+ vm1 = getVM(1);
+ vm2 = getVM(2);
+ vm3 = getVM(3);
+ vm4 = getVM(4);
+ vm5 = getVM(5);
+ vm6 = getVM(6);
+ vm7 = getVM(7);
+
+ // Stopping the gateway closed the region, which causes this exception to get logged
+ addIgnoredException(RegionDestroyedException.class);
+ }
+
+ @Test
+ public void testStartPauseResumeSerialGatewaySenderUpdateAttributes() throws Exception {
+ int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
+ int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));
+
+ for (VM vm : toArray(vm2, vm3)) {
+ vm.invoke(() -> {
+ createCache(nyPort);
+ createReceiver();
+ });
+ }
+
+ vm4.invoke(() -> createCache(lnPort));
+ vm5.invoke(() -> createCache(lnPort));
+ vm6.invoke(() -> createCache(lnPort));
+ vm7.invoke(() -> createCache(lnPort));
+
+ vm4.invoke(() -> createSenderInVm4());
+ vm5.invoke(() -> createSenderInVm5());
+
+ vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
+ vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));
+
+ vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+
+ for (VM vm : toArray(vm4, vm5)) {
+ vm.invoke(() -> startSender("ln"));
+ }
+
+ vm4.invoke(() -> pauseSender("ln"));
+ vm5.invoke(() -> pauseSender("ln"));
+
+ vm4.invoke(() -> validateSenderPausedState("ln"));
+ vm5.invoke(() -> validateSenderPausedState("ln"));
+
+ vm4.invoke(() -> doPuts(className + "_RR", 1000));
+
+ updateBatchSize(50);
+ updateBatchTimeInterval(200);
+
+ vm4.invoke(() -> resumeSender("ln"));
+ vm5.invoke(() -> resumeSender("ln"));
+
+ vm4.invoke(() -> validateSenderResumedState("ln"));
+ vm5.invoke(() -> validateSenderResumedState("ln"));
+
+ checkBatchSize(50);
+ checkBatchTimeInterval(200);
+
+ vm4.invoke(() -> validateQueueContents("ln", 0));
+ vm5.invoke(() -> validateQueueContents("ln", 0));
+
+ vm2.invoke(() -> validateRegionSize(className + "_RR", 1000));
+ vm3.invoke(() -> validateRegionSize(className + "_RR", 1000));
+ }
+
+ @Test
+ public void testSerialGatewaySenderUpdateAttributesWhilePutting() throws Exception {
+ int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
+ int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));
+
+ for (VM vm : toArray(vm2, vm3)) {
+ vm.invoke(() -> {
+ createCache(nyPort);
+ createReceiver();
+ });
+ }
+
+ vm4.invoke(() -> createCache(lnPort));
+ vm5.invoke(() -> createCache(lnPort));
+ vm6.invoke(() -> createCache(lnPort));
+ vm7.invoke(() -> createCache(lnPort));
+
+ vm4.invoke(() -> createSenderInVm4());
+ vm5.invoke(() -> createSenderInVm5());
+
+ vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
+ vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));
+
+ vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+
+ for (VM vm : toArray(vm4, vm5)) {
+ vm.invoke(() -> startSender("ln"));
+ }
+
+ // Do some puts from both vm4 and vm5 while restarting a sender
+ AsyncInvocation doPutsInVm4 =
+ vm4.invokeAsync(() -> doPuts(className + "_RR", 1000));
+
+ updateBatchSize(50);
+ updateBatchTimeInterval(200);
+
+ doPutsInVm4.await();
+
+ checkBatchSize(50);
+ checkBatchTimeInterval(200);
+
+ vm4.invoke(() -> validateQueueContents("ln", 0));
+ vm5.invoke(() -> validateQueueContents("ln", 0));
+
+ vm2.invoke(() -> validateRegionSize(className + "_RR", 1000));
+ vm3.invoke(() -> validateRegionSize(className + "_RR", 1000));
+ }
+
+ @Test
+ public void testSerialGatewaySenderUpdateGatewayEventFiltersWhilePutting() throws Exception {
+ int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1));
+ int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort));
+
+ List<GatewayEventFilter> filters = new ArrayList<>();
+ filters.add(new MyGatewayEventFilter_AfterAck());
+ filters.add(new PDXGatewayEventFilter());
+
+ for (VM vm : toArray(vm2, vm3)) {
+ vm.invoke(() -> {
+ createCache(nyPort);
+ createReceiver();
+ });
+ }
+
+ vm4.invoke(() -> createCache(lnPort));
+ vm5.invoke(() -> createCache(lnPort));
+ vm6.invoke(() -> createCache(lnPort));
+ vm7.invoke(() -> createCache(lnPort));
+
+ vm4.invoke(() -> createSenderInVm4());
+ vm5.invoke(() -> createSenderInVm5());
+
+ vm2.invoke(() -> createReplicatedRegion(className + "_RR", null));
+ vm3.invoke(() -> createReplicatedRegion(className + "_RR", null));
+
+ vm4.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm5.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm6.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+ vm7.invoke(() -> createReplicatedRegion(className + "_RR", "ln"));
+
+ for (VM vm : toArray(vm4, vm5)) {
+ vm.invoke(() -> startSender("ln"));
+ }
+
+ // Do some puts from both vm4 and vm5 while restarting a sender
+ AsyncInvocation doPutsInVm4 =
+ vm4.invokeAsync(() -> doPuts(className + "_RR", 5000));
+
+ updateBatchSize(40);
+ updateGatewayEventFilters(filters);
+
+ doPutsInVm4.await();
+
+ checkBatchSize(40);
+
+ vm4.invoke(() -> validateQueueContents("ln", 0));
+ vm5.invoke(() -> validateQueueContents("ln", 0));
+
+ vm2.invoke(() -> validateRegionSize(className + "_RR", 5000));
+ vm3.invoke(() -> validateRegionSize(className + "_RR", 5000));
+ }
+
+ protected boolean isOffHeap() {
+ return false;
+ }
+
+ protected void createSenderInVm4() throws IOException {
+ createSender("ln", 2, true, true, numDispatchers, DEFAULT_ORDER_POLICY);
+ }
+
+ protected void createSenderInVm5() throws IOException {
+ createSender("ln", 2, true, true, numDispatchers, DEFAULT_ORDER_POLICY);
+ }
+
+ protected final void createSender(String id,
+ int remoteDsId,
+ boolean isPersistent,
+ boolean isManualStart,
+ int numDispatchers,
+ OrderPolicy policy) throws IOException {
+ try (IgnoredException ie = addIgnoredException("Could not connect")) {
+ File persistentDirectory =
+ temporaryFolder.newFolder(id + "_disk_" + currentTimeMillis() + "_" + getCurrentVMNum());
+ DiskStoreFactory diskStoreFactory = getCache().createDiskStoreFactory();
+ File[] dirs = new File[] {persistentDirectory};
+
+ InternalGatewaySenderFactory gatewaySenderFactory =
+ (InternalGatewaySenderFactory) getCache().createGatewaySenderFactory();
+
+ gatewaySenderFactory.setParallel(false);
+ gatewaySenderFactory.setMaximumQueueMemory(100);
+ gatewaySenderFactory.setBatchSize(10);
+ gatewaySenderFactory.setBatchConflationEnabled(false);
+ gatewaySenderFactory.setManualStart(isManualStart);
+ gatewaySenderFactory.setDispatcherThreads(numDispatchers);
+ gatewaySenderFactory.setOrderPolicy(policy);
+ gatewaySenderFactory.setSocketBufferSize(DEFAULT_SOCKET_BUFFER_SIZE);
+
+ if (isPersistent) {
+ gatewaySenderFactory.setPersistenceEnabled(true);
+ gatewaySenderFactory.setDiskStoreName(
+ diskStoreFactory.setDiskDirs(dirs).create(id).getName());
+ } else {
+ DiskStore store = diskStoreFactory.setDiskDirs(dirs).create(id);
+ gatewaySenderFactory.setDiskStoreName(store.getName());
+ }
+
+ gatewaySenderFactory.create(id, remoteDsId);
+ }
+ }
+
+ private Properties getDistributedSystemProperties(int locatorPort) {
+ Properties props = getDistributedSystemProperties();
+ props.setProperty(MCAST_PORT, "0");
+ props.setProperty(LOCATORS, "localhost[" + locatorPort + "]");
+ return props;
+ }
+
+ private void createCache(int locatorPort) {
+ getCache(getDistributedSystemProperties(locatorPort));
+ }
+
+ private void createReplicatedRegion(String regionName, String senderIds) {
+ try (IgnoredException ie1 = addIgnoredException(ForceReattemptException.class);
+ IgnoredException ie2 = addIgnoredException(GatewaySenderException.class);
+ IgnoredException ie3 = addIgnoredException(InterruptedException.class)) {
+ RegionFactory regionFactory = getCache().createRegionFactory(REPLICATE);
+
+ if (senderIds != null) {
+ StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
+ while (tokenizer.hasMoreTokens()) {
+ String senderId = tokenizer.nextToken();
+ regionFactory.addGatewaySenderId(senderId);
+ }
+ }
+
+ regionFactory.setDataPolicy(DataPolicy.REPLICATE);
+ regionFactory.setScope(Scope.DISTRIBUTED_ACK);
+ regionFactory.setOffHeap(isOffHeap());
+
+ regionFactory.create(regionName);
+ }
+ }
+
+ private void doPuts(String regionName, int count) {
+ try (IgnoredException ie1 = addIgnoredException(GatewaySenderException.class);
+ IgnoredException ie2 = addIgnoredException(InterruptedException.class)) {
+ Region<Number, String> region = getCache().getRegion(SEPARATOR + regionName);
+ for (int i = 0; i < count; i++) {
+ region.put(i, "Value_" + i);
+ }
+ }
+ }
+
+ private void doPuts(String regionName, int from, int count) {
+ Region<Number, String> region = getCache().getRegion(SEPARATOR + regionName);
+ for (int i = from; i < count; i++) {
+ region.put(i, "Value_" + i);
+ }
+ }
+
+ private int createFirstLocatorWithDSId(int systemId) {
+ stopOldLocator();
+ int locatorPort = getRandomAvailableTCPPort();
+ startLocator(systemId, locatorPort, locatorPort, -1, true);
+ return locatorPort;
+ }
+
+ private int createFirstRemoteLocator(int systemId, int remoteLocatorPort) {
+ stopOldLocator();
+ int locatorPort = getRandomAvailableTCPPort();
+ startLocator(systemId, locatorPort, locatorPort, remoteLocatorPort, true);
+ return locatorPort;
+ }
+
+ private void startLocator(int systemId, int locatorPort, int startLocatorPort,
+ int remoteLocatorPort, boolean startServerLocator) {
+ Properties props = getDistributedSystemProperties();
+
+ props.setProperty(DISTRIBUTED_SYSTEM_ID, String.valueOf(systemId));
+ props.setProperty(MCAST_PORT, "0");
+ props.setProperty(LOCATORS, "localhost[" + locatorPort + "]");
+ props.setProperty(START_LOCATOR, "localhost[" + startLocatorPort + "],server="
+ + startServerLocator + ",peer=true,hostname-for-clients=localhost");
+
+ if (remoteLocatorPort != -1) {
+ props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocatorPort + "]");
+ }
+
+ // Start start the locator with a LOCATOR_DM_TYPE and not a NORMAL_DM_TYPE
+ System.setProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE, "true");
+ try {
+ getSystem(props);
+ } finally {
+ System.clearProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE);
+ }
+ }
+
+ private void stopOldLocator() {
+ if (Locator.hasLocator()) {
+ Locator.getLocator().stop();
+ }
+ }
+
+ private InternalGatewaySender findInternalGatewaySender(String senderId) {
+ return (InternalGatewaySender) findGatewaySender(senderId, true);
+ }
+
+ private GatewaySender findGatewaySender(String senderId) {
+ return findGatewaySender(senderId, true);
+ }
+
+ private GatewaySender findGatewaySender(String senderId, boolean assertNotNull) {
+ Set<GatewaySender> senders = getCache().getGatewaySenders();
+ GatewaySender sender = null;
+ for (GatewaySender s : senders) {
+ if (s.getId().equals(senderId)) {
+ sender = s;
+ break;
+ }
+ }
+
+ if (assertNotNull) {
+ assertThat(sender).isNotNull();
+ }
+
+ return sender;
+ }
+
+ private void startSender(String senderId) {
+ try (IgnoredException ie1 = addIgnoredException("Could not connect");
+ IgnoredException ie2 = addIgnoredException(ForceReattemptException.class);
+ IgnoredException ie3 = addIgnoredException(InterruptedException.class)) {
+ findGatewaySender(senderId).start();
+ }
+ }
+
+ private void pauseSender(String senderId) {
+ try (IgnoredException ie1 = addIgnoredException("Could not connect");
+ IgnoredException ie2 = addIgnoredException(ForceReattemptException.class)) {
+ InternalGatewaySender sender = findInternalGatewaySender(senderId);
+ sender.pause();
+ sender.getEventProcessor().waitForDispatcherToPause();
+ }
+ }
+
+ private void resumeSender(String senderId) {
+ try (IgnoredException ie1 = addIgnoredException("Could not connect");
+ IgnoredException ie2 = addIgnoredException(ForceReattemptException.class)) {
+ findGatewaySender(senderId).resume();
+ }
+ }
+
+ private void stopSender(String senderId) {
+ try (IgnoredException ie1 = addIgnoredException("Could not connect");
+ IgnoredException ie2 = addIgnoredException(ForceReattemptException.class)) {
+ InternalGatewaySender sender = findInternalGatewaySender(senderId);
+
+ AbstractGatewaySenderEventProcessor eventProcessor = sender.getEventProcessor();
+
+ sender.stop();
+
+ if (eventProcessor instanceof ConcurrentSerialGatewaySenderEventProcessor) {
+ ConcurrentSerialGatewaySenderEventProcessor concurrentEventProcessor =
+ (ConcurrentSerialGatewaySenderEventProcessor) eventProcessor;
+ Set<RegionQueue> queues = concurrentEventProcessor.getQueues();
+ for (RegionQueue queue : queues) {
+ if (queue instanceof SerialGatewaySenderQueue) {
+ assertThat(((SerialGatewaySenderQueue) queue).isRemovalThreadAlive())
+ .isFalse();
+ }
+ }
+ }
+ }
+ }
+
+ private int createReceiver() throws IOException {
+ int receiverPort = getRandomAvailableTCPPort();
+
+ GatewayReceiverFactory gatewayReceiverFactory = getCache().createGatewayReceiverFactory();
+ gatewayReceiverFactory.setStartPort(receiverPort);
+ gatewayReceiverFactory.setEndPort(receiverPort);
+ gatewayReceiverFactory.setManualStart(true);
+
+ GatewayReceiver receiver = gatewayReceiverFactory.create();
+ receiver.start();
+
+ return receiverPort;
+ }
+
+ private void validateRegionSize(String regionName, int regionSize) {
+ try (IgnoredException ie1 = addIgnoredException(CacheClosedException.class);
+ IgnoredException ie2 = addIgnoredException(ForceReattemptException.class)) {
+ Region region = getCache().getRegion(SEPARATOR + regionName);
+
+ await()
+ .untilAsserted(() -> {
+ assertThat(region.keySet()).hasSize(regionSize);
+ });
+ }
+ }
+
+ private void validateQueueContents(String senderId, int regionSize) {
+ try (IgnoredException ie1 = addIgnoredException(GatewaySenderException.class);
+ IgnoredException ie2 = addIgnoredException(InterruptedException.class)) {
+ InternalGatewaySender sender = findInternalGatewaySender(senderId);
+
+ if (sender.isParallel()) {
+ RegionQueue regionQueue = sender.getQueues().toArray(new RegionQueue[1])[0];
+ await()
+ .untilAsserted(() -> {
+ assertThat(regionQueue.size())
+ .isEqualTo(regionSize);
+ });
+
+ } else {
+ Set<RegionQueue> queues = sender.getQueues();
+ await()
+ .untilAsserted(() -> {
+ int size = 0;
+ for (RegionQueue queue : queues) {
+ size += queue.size();
+ }
+ assertThat(size)
+ .isEqualTo(regionSize);
+ });
+ }
+ }
+ }
+
+ private void validateSenderPausedState(String senderId) {
+ GatewaySender sender = findGatewaySender(senderId);
+
+ assertThat(sender.isPaused())
+ .isTrue();
+ }
+
+ private void validateSenderResumedState(String senderId) {
+ GatewaySender sender = findGatewaySender(senderId);
+
+ assertThat(sender.isPaused())
+ .isFalse();
+ assertThat(sender.isRunning())
+ .isTrue();
+ }
+
+ private void updateBatchSize(int batchsize) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchSize(batchsize);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchSize(batchsize);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void updateBatchTimeInterval(int batchTimeInterval) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchTimeInterval(batchTimeInterval);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setBatchTimeInterval(batchTimeInterval);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void updateGroupTransactionEvents(boolean groupTransactionEvents) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGroupTransactionEvents(groupTransactionEvents);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGroupTransactionEvents(groupTransactionEvents);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void updateGatewayEventFilters(List<GatewayEventFilter> filters) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGatewayEventFilters(filters);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ boolean paused = false;
+ if (sender.isRunning() && !sender.isPaused()) {
+ sender.pause();
+ paused = true;
+ }
+ sender.setGatewayEventFilters(filters);
+ if (paused) {
+ sender.resume();
+ }
+ });
+ }
+
+ private void checkBatchSize(int batchsize) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchSize()).isEqualTo(batchsize);
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchSize()).isEqualTo(batchsize);
+ });
+ }
+
+ private void checkBatchTimeInterval(int batchTimeInterval) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval);
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval);
+ });
+ }
+
+ private void checkGroupTransactionEvents(boolean groupTransactionEvents) {
+ vm4.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents);
+ });
+ vm5.invoke(() -> {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln");
+ assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents);
+ });
+
+ }
+
+
+ public static class MyGatewayEventFilter implements GatewayEventFilter, Serializable {
+
+ String Id = "MyGatewayEventFilter";
+
+ boolean beforeEnqueueInvoked;
+ boolean beforeTransmitInvoked;
+ boolean afterAckInvoked;
+
+ public MyGatewayEventFilter() {}
+
+ @Override
+ public boolean beforeEnqueue(GatewayQueueEvent event) {
+ this.beforeEnqueueInvoked = true;
+ return !((Integer) event.getKey() >= 500 && (Integer) event.getKey() < 600);
+ }
+
+ @Override
+ public boolean beforeTransmit(GatewayQueueEvent event) {
+ this.beforeTransmitInvoked = true;
+ return !((Integer) event.getKey() >= 600 && (Integer) event.getKey() < 700);
+ }
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
+
+ public String toString() {
+ return Id;
+ }
+
+ @Override
+ public void afterAcknowledgement(GatewayQueueEvent event) {
+ this.afterAckInvoked = true;
+ // TODO Auto-generated method stub
+ }
+
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof MyGatewayEventFilter))
+ return false;
+ MyGatewayEventFilter filter = (MyGatewayEventFilter) obj;
+ return this.Id.equals(filter.Id);
+ }
+ }
+
+ public static class MyGatewayEventFilter_AfterAck implements GatewayEventFilter, Serializable {
+
+ String Id = "MyGatewayEventFilter_AfterAck";
+
+ ConcurrentSkipListSet<Integer> ackList = new ConcurrentSkipListSet<Integer>();
+
+ public MyGatewayEventFilter_AfterAck() {}
+
+ @Override
+ public boolean beforeEnqueue(GatewayQueueEvent event) {
+ return true;
+ }
+
+ @Override
+ public boolean beforeTransmit(GatewayQueueEvent event) {
+ return true;
+ }
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
+
+ public String toString() {
+ return Id;
+ }
+
+ @Override
+ public void afterAcknowledgement(GatewayQueueEvent event) {
+ ackList.add((Integer) event.getKey());
+ }
+
+ public Set getAckList() {
+ return ackList;
+ }
+
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof MyGatewayEventFilter))
+ return false;
+ MyGatewayEventFilter filter = (MyGatewayEventFilter) obj;
+ return this.Id.equals(filter.Id);
+ }
+ }
+
+ public static class PDXGatewayEventFilter implements GatewayEventFilter, Serializable {
+
+ String Id = "PDXGatewayEventFilter";
+
+ public int beforeEnqueueInvoked;
+ public int beforeTransmitInvoked;
+ public int afterAckInvoked;
+
+ public PDXGatewayEventFilter() {}
+
+ @Override
+ public boolean beforeEnqueue(GatewayQueueEvent event) {
+ System.out.println("Invoked enqueue for " + event);
+ this.beforeEnqueueInvoked++;
+ return true;
+ }
+
+ @Override
+ public boolean beforeTransmit(GatewayQueueEvent event) {
+ System.out.println("Invoked transmit for " + event);
+ this.beforeTransmitInvoked++;
+ return true;
+ }
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
+
+ public String toString() {
+ return Id;
+ }
+
+ @Override
+ public void afterAcknowledgement(GatewayQueueEvent event) {
+ System.out.println("Invoked afterAck for " + event);
+ this.afterAckInvoked++;
+ // TODO Auto-generated method stub
+ }
+
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof WANTestBase.MyGatewayEventFilter))
+ return false;
+ MyGatewayEventFilter filter = (MyGatewayEventFilter) obj;
+ return this.Id.equals(filter.Id);
+ }
+ }
+
+
+}
diff --git a/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/wancommand/AlterGatewaySenderCommandDUnitTest.java b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/wancommand/AlterGatewaySenderCommandDUnitTest.java
new file mode 100644
index 0000000..14cc30b
--- /dev/null
+++ b/geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/wancommand/AlterGatewaySenderCommandDUnitTest.java
@@ -0,0 +1,488 @@
+/*
+ * 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.internal.cache.wan.wancommand;
+
+import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
+import static org.apache.geode.distributed.ConfigurationProperties.NAME;
+import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Properties;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.wan.GatewayEventFilter;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.distributed.internal.InternalLocator;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.test.dunit.IgnoredException;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.WanTest;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+
+@Category({WanTest.class})
+public class AlterGatewaySenderCommandDUnitTest {
+
+ public static final String CREATE =
+ "create gateway-sender --id=sender1 --remote-distributed-system-id=2";
+ public static final String CREATE_PARALLEL =
+ "create gateway-sender --id=sender1P --remote-distributed-system-id=2 --parallel=true";
+ public static final String DESTROY = "destroy gateway-sender --id=sender1";
+ public static final String DESTROY_PARALLEL = "destroy gateway-sender --id=sender1P";
+
+
+ @ClassRule
+ public static ClusterStartupRule clusterStartupRule = new ClusterStartupRule();
+
+ @Rule
+ public GfshCommandRule gfsh = new GfshCommandRule();
+
+ private static MemberVM locatorSite1;
+ private static MemberVM server1;
+ private static MemberVM server2;
+ private final IgnoredException exln = IgnoredException
+ .addIgnoredException("could not get remote locator information for remote site");
+ private final IgnoredException exln1 = IgnoredException
+ .addIgnoredException("Connection reset");
+ private final IgnoredException exln2 = IgnoredException
+ .addIgnoredException("Broken pipe");
+ private final IgnoredException exln3 = IgnoredException
+ .addIgnoredException("Connection refused");
+ private final IgnoredException exln4 = IgnoredException
+ .addIgnoredException("Unexpected IOException");
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(NAME, "happylocator");
+ props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + 1);
+ locatorSite1 = clusterStartupRule.startLocatorVM(0, props);
+
+ props.setProperty(NAME, "happyserver1");
+ server1 = clusterStartupRule.startServerVM(1, props, locatorSite1.getPort());
+
+ props.setProperty(NAME, "happyserver2");
+ server2 = clusterStartupRule.startServerVM(2, props, locatorSite1.getPort());
+
+ props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + 2);
+ props.setProperty(NAME, "happyremotelocator");
+ props.setProperty(REMOTE_LOCATORS, "localhost[" + locatorSite1.getPort() + "]");
+ clusterStartupRule.startLocatorVM(3, props);
+ }
+
+ @Before
+ public void before() throws Exception {
+ gfsh.connectAndVerify(locatorSite1);
+ }
+
+ @After
+ public void after() {
+ gfsh.executeAndAssertThat(DESTROY + " --if-exists").statusIsSuccess();
+ gfsh.executeAndAssertThat(DESTROY_PARALLEL + " --if-exists").statusIsSuccess();
+ exln.remove();
+ exln1.remove();
+ exln2.remove();
+ exln3.remove();
+ exln4.remove();
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderWithDefault() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(GatewaySender.DEFAULT_BATCH_SIZE);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(GatewaySender.DEFAULT_ALERT_THRESHOLD);
+ assertThat(sender.getDispatcherThreads()).isEqualTo(GatewaySender.DEFAULT_DISPATCHER_THREADS);
+ });
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderAndAlterBatchSize() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=200 --alert-threshold=100")
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ });
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderAndInvalidAlterBatchSize() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=-10 --alert-threshold=100")
+ .statusIsError();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(GatewaySender.DEFAULT_BATCH_SIZE);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(GatewaySender.DEFAULT_ALERT_THRESHOLD);
+ });
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderAndAlterBatchSizeCheckConfig() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=200 --alert-threshold=100")
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ });
+
+ // verify that server1's event queue has the default value
+ server2.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ });
+
+ locatorSite1.invoke(() -> {
+ InternalLocator locator = ClusterStartupRule.getLocator();
+ assertThat(locator).isNotNull();
+ String xml = locator.getConfigurationPersistenceService().getConfiguration("cluster")
+ .getCacheXmlContent();
+ assertThat(xml).contains("batch-size=\"200\"");
+ assertThat(xml).contains("alert-threshold=\"100\"");
+ });
+
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderAndChangeGroupTransaction() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ gfsh.executeAndAssertThat("alter gateway-sender --id=sender1 --group-transaction-events=true")
+ .statusIsError()
+ .containsOutput("alter-gateway-sender cannot be performed for --group-transaction-events");
+ }
+
+
+ @Test
+ public void testCreateSerialGatewaySenderAndAlterBatchSizeServerDown() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ server1.stop(false);
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=200 --alert-threshold=100")
+ .statusIsSuccess();
+
+ Properties props = new Properties();
+ props.setProperty(NAME, "happyserver1");
+ props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + 1);
+ server1 = clusterStartupRule.startServerVM(1, props, locatorSite1.getPort());
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ });
+
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderAndAlterEventFiters() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=200 --alert-threshold=100 --gateway-event-filter="
+ + MyGatewayEventFilter.class.getName())
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ assertThat(sender.getGatewayEventFilters().get(0).beforeEnqueue(null)).isTrue();
+ assertThat(sender.getGatewayEventFilters().size()).isEqualTo(1);
+
+ });
+ server2.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ assertThat(sender.getGatewayEventFilters().get(0).beforeEnqueue(null)).isTrue();
+ assertThat(sender.getGatewayEventFilters().size()).isEqualTo(1);
+ });
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=2000 --alert-threshold=1000")
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(2000);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(1000);
+ assertThat(sender.getGatewayEventFilters().get(0).beforeEnqueue(null)).isTrue();
+ assertThat(sender.getGatewayEventFilters().size()).isEqualTo(1);
+
+ });
+ server2.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(2000);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(1000);
+ assertThat(sender.getGatewayEventFilters().get(0).beforeEnqueue(null)).isTrue();
+ assertThat(sender.getGatewayEventFilters().size()).isEqualTo(1);
+ });
+
+ }
+
+ @Test
+ public void testCreateSerialGatewaySenderAndAlterEventFitersAndRemove() throws Exception {
+ gfsh.executeAndAssertThat(CREATE).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1\" created on \"happyserver1\"",
+ "GatewaySender \"sender1\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1");
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=200 --alert-threshold=100 --gateway-event-filter="
+ + MyGatewayEventFilter.class.getName())
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ assertThat(sender.getGatewayEventFilters().get(0).beforeEnqueue(null)).isTrue();
+ assertThat(sender.getGatewayEventFilters().size()).isEqualTo(1);
+
+ });
+ server2.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ assertThat(sender.getGatewayEventFilters().get(0).beforeEnqueue(null)).isTrue();
+ assertThat(sender.getGatewayEventFilters().size()).isEqualTo(1);
+ });
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1 --batch-size=111 --alert-threshold=55 --gateway-event-filter")
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(111);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(55);
+ assertThat(sender.getGatewayEventFilters().isEmpty()).isTrue();
+
+ });
+ server2.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1");
+ assertThat(sender.getBatchSize()).isEqualTo(111);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(55);
+ assertThat(sender.getGatewayEventFilters().isEmpty()).isTrue();
+ });
+ }
+
+
+ @Test
+ public void testCreateParallelGatewaySenderAndAlterBatchSize() throws Exception {
+ gfsh.executeAndAssertThat(CREATE_PARALLEL).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1P\" created on \"happyserver1\"",
+ "GatewaySender \"sender1P\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1P");
+
+ gfsh.executeAndAssertThat(
+ "alter gateway-sender --id=sender1P --batch-size=200 --alert-threshold=100")
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1P");
+ assertThat(sender.getBatchSize()).isEqualTo(200);
+ assertThat(sender.getBatchTimeInterval())
+ .isEqualTo(GatewaySender.DEFAULT_BATCH_TIME_INTERVAL);
+ assertThat(sender.getAlertThreshold()).isEqualTo(100);
+ });
+ }
+
+ @Test
+ public void testCreateParallelGatewaySenderAndChangeGroupTransaction() throws Exception {
+ gfsh.executeAndAssertThat(CREATE_PARALLEL).statusIsSuccess()
+ .doesNotContainOutput("Did not complete waiting")
+ .hasTableSection()
+ .hasColumn("Message")
+ .containsExactly("GatewaySender \"sender1P\" created on \"happyserver1\"",
+ "GatewaySender \"sender1P\" created on \"happyserver2\"");
+
+ gfsh.executeAndAssertThat("list gateways").statusIsSuccess()
+ .containsOutput("sender1P");
+
+ gfsh.executeAndAssertThat("alter gateway-sender --id=sender1P --group-transaction-events=true")
+ .statusIsSuccess();
+
+ // verify that server1's event queue has the default value
+ server1.invoke(() -> {
+ InternalCache cache = ClusterStartupRule.getCache();
+ GatewaySender sender = cache.getGatewaySender("sender1P");
+ assertThat(sender.mustGroupTransactionEvents()).isTrue();
+ });
+ }
+
+ public static class MyGatewayEventFilter implements GatewayEventFilter {
+ @Override
+ public void afterAcknowledgement(GatewayQueueEvent event) {}
+
+ @Override
+ public boolean beforeEnqueue(GatewayQueueEvent event) {
+ return true;
+ }
+
+ @Override
+ public boolean beforeTransmit(GatewayQueueEvent event) {
+ return true;
+ }
+
+ }
+}
diff --git a/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/GatewaySenderFactoryImpl.java b/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/GatewaySenderFactoryImpl.java
index 0704602..e78f750 100644
--- a/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/GatewaySenderFactoryImpl.java
+++ b/geode-wan/src/main/java/org/apache/geode/cache/wan/internal/GatewaySenderFactoryImpl.java
@@ -68,19 +68,19 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
@Override
public GatewaySenderFactory setParallel(boolean isParallel) {
- this.attrs.isParallel = isParallel;
+ this.attrs.setParallel(isParallel);
return this;
}
@Override
public GatewaySenderFactory setGroupTransactionEvents(boolean groupTransactionEvents) {
- this.attrs.groupTransactionEvents = groupTransactionEvents;
+ this.attrs.setGroupTransactionEvents(groupTransactionEvents);
return this;
}
@Override
public GatewaySenderFactory setForInternalUse(boolean isForInternalUse) {
- this.attrs.isForInternalUse = isForInternalUse;
+ this.attrs.setForInternalUse(isForInternalUse);
return this;
}
@@ -104,108 +104,108 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
@Override
public GatewaySenderFactory setSocketBufferSize(int socketBufferSize) {
- this.attrs.socketBufferSize = socketBufferSize;
+ this.attrs.setSocketBufferSize(socketBufferSize);
return this;
}
@Override
public GatewaySenderFactory setSocketReadTimeout(int socketReadTimeout) {
- this.attrs.socketReadTimeout = socketReadTimeout;
+ this.attrs.setSocketReadTimeout(socketReadTimeout);
return this;
}
@Override
public GatewaySenderFactory setDiskStoreName(String diskStoreName) {
- this.attrs.diskStoreName = diskStoreName;
+ this.attrs.setDiskStoreName(diskStoreName);
return this;
}
@Override
public GatewaySenderFactory setMaximumQueueMemory(int maximumQueueMemory) {
- this.attrs.maximumQueueMemory = maximumQueueMemory;
+ this.attrs.setMaximumQueueMemory(maximumQueueMemory);
return this;
}
@Override
public GatewaySenderFactory setBatchSize(int batchSize) {
- this.attrs.batchSize = batchSize;
+ this.attrs.setBatchSize(batchSize);
return this;
}
@Override
public GatewaySenderFactory setBatchTimeInterval(int batchTimeInterval) {
- this.attrs.batchTimeInterval = batchTimeInterval;
+ this.attrs.setBatchTimeInterval(batchTimeInterval);
return this;
}
@Override
public GatewaySenderFactory setBatchConflationEnabled(boolean enableBatchConflation) {
- this.attrs.isBatchConflationEnabled = enableBatchConflation;
+ this.attrs.setBatchConflationEnabled(enableBatchConflation);
return this;
}
@Override
public GatewaySenderFactory setPersistenceEnabled(boolean enablePersistence) {
- this.attrs.isPersistenceEnabled = enablePersistence;
+ this.attrs.setPersistenceEnabled(enablePersistence);
return this;
}
@Override
public GatewaySenderFactory setAlertThreshold(int threshold) {
- this.attrs.alertThreshold = threshold;
+ this.attrs.setAlertThreshold(threshold);
return this;
}
@Override
public GatewaySenderFactory setManualStart(boolean start) {
- this.attrs.manualStart = start;
+ this.attrs.setManualStart(start);
return this;
}
@Override
public GatewaySenderFactory setLocatorDiscoveryCallback(LocatorDiscoveryCallback locCallback) {
- this.attrs.locatorDiscoveryCallback = locCallback;
+ this.attrs.setLocatorDiscoveryCallback(locCallback);
return this;
}
@Override
public GatewaySenderFactory setDiskSynchronous(boolean isSynchronous) {
- this.attrs.isDiskSynchronous = isSynchronous;
+ this.attrs.setDiskSynchronous(isSynchronous);
return this;
}
@Override
public GatewaySenderFactory setDispatcherThreads(int numThreads) {
- if ((numThreads > 1) && this.attrs.policy == null) {
- this.attrs.policy = GatewaySender.DEFAULT_ORDER_POLICY;
+ if ((numThreads > 1) && this.attrs.getOrderPolicy() == null) {
+ this.attrs.setOrderPolicy(GatewaySender.DEFAULT_ORDER_POLICY);
}
- this.attrs.dispatcherThreads = numThreads;
+ this.attrs.setDispatcherThreads(numThreads);
return this;
}
@Override
public GatewaySenderFactory setParallelFactorForReplicatedRegion(int parallel) {
- this.attrs.parallelism = parallel;
- this.attrs.policy = GatewaySender.DEFAULT_ORDER_POLICY;
+ this.attrs.setParallelism(parallel);
+ this.attrs.setOrderPolicy(GatewaySender.DEFAULT_ORDER_POLICY);
return this;
}
@Override
public GatewaySenderFactory setOrderPolicy(OrderPolicy policy) {
- this.attrs.policy = policy;
+ this.attrs.setOrderPolicy(policy);
return this;
}
@Override
public GatewaySenderFactory setBucketSorted(boolean isBucketSorted) {
- this.attrs.isBucketSorted = isBucketSorted;
+ this.attrs.setBucketSorted(isBucketSorted);
return this;
}
@Override
public GatewaySenderFactory setEnforceThreadsConnectSameReceiver(
boolean enforceThreadsConnectSameReceiver) {
- this.attrs.enforceThreadsConnectSameReceiver = enforceThreadsConnectSameReceiver;
+ this.attrs.setEnforceThreadsConnectSameReceiver(enforceThreadsConnectSameReceiver);
return this;
}
@@ -224,8 +224,8 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
String.format("GatewaySender %s cannot be created with remote DS Id less than 0. ",
id));
}
- this.attrs.id = id;
- this.attrs.remoteDs = remoteDSId;
+ this.attrs.setId(id);
+ this.attrs.setRemoteDs(remoteDSId);
GatewaySender sender = null;
if (this.attrs.getDispatcherThreads() <= 0) {
@@ -245,7 +245,7 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
"{} cannot configure socket read timeout of {} milliseconds because it is less than the minimum of {} milliseconds. The default will be used instead.",
new Object[] {"GatewaySender " + id, this.attrs.getSocketReadTimeout(),
GatewaySender.MINIMUM_SOCKET_READ_TIMEOUT});
- this.attrs.socketReadTimeout = GatewaySender.MINIMUM_SOCKET_READ_TIMEOUT;
+ this.attrs.setSocketReadTimeout(GatewaySender.MINIMUM_SOCKET_READ_TIMEOUT);
}
// Log a warning if the old system property is set.
@@ -297,7 +297,7 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
id));
}
if (this.attrs.getOrderPolicy() == null && this.attrs.getDispatcherThreads() > 1) {
- this.attrs.policy = GatewaySender.DEFAULT_ORDER_POLICY;
+ this.attrs.setOrderPolicy(GatewaySender.DEFAULT_ORDER_POLICY);
}
if (this.cache instanceof GemFireCacheImpl) {
sender = new SerialGatewaySenderImpl(cache, statisticsClock, attrs);
@@ -315,7 +315,7 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
@Override
public GatewaySender create(String id) {
- this.attrs.id = id;
+ this.attrs.setId(id);
GatewaySender sender = null;
if (this.attrs.getDispatcherThreads() <= 0) {
@@ -345,7 +345,7 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
}
} else {
if (this.attrs.getOrderPolicy() == null && this.attrs.getDispatcherThreads() > 1) {
- this.attrs.policy = GatewaySender.DEFAULT_ORDER_POLICY;
+ this.attrs.setOrderPolicy(GatewaySender.DEFAULT_ORDER_POLICY);
}
if (this.cache instanceof GemFireCacheImpl) {
sender = new SerialGatewaySenderImpl(cache, statisticsClock, attrs);
@@ -363,48 +363,48 @@ public class GatewaySenderFactoryImpl implements InternalGatewaySenderFactory {
@Override
public GatewaySenderFactory removeGatewayEventFilter(GatewayEventFilter filter) {
- this.attrs.eventFilters.remove(filter);
+ this.attrs.getGatewayEventFilters().remove(filter);
return this;
}
@Override
public GatewaySenderFactory removeGatewayTransportFilter(GatewayTransportFilter filter) {
- this.attrs.transFilters.remove(filter);
+ this.attrs.getGatewayTransportFilters().remove(filter);
return this;
}
@Override
public GatewaySenderFactory setGatewayEventSubstitutionFilter(
GatewayEventSubstitutionFilter filter) {
- this.attrs.eventSubstitutionFilter = filter;
+ this.attrs.setEventSubstitutionFilter(filter);
return this;
}
@Override
public void configureGatewaySender(GatewaySender senderCreation) {
- this.attrs.isParallel = senderCreation.isParallel();
- this.attrs.manualStart = senderCreation.isManualStart();
- this.attrs.socketBufferSize = senderCreation.getSocketBufferSize();
- this.attrs.socketReadTimeout = senderCreation.getSocketReadTimeout();
- this.attrs.isBatchConflationEnabled = senderCreation.isBatchConflationEnabled();
- this.attrs.batchSize = senderCreation.getBatchSize();
- this.attrs.batchTimeInterval = senderCreation.getBatchTimeInterval();
- this.attrs.isPersistenceEnabled = senderCreation.isPersistenceEnabled();
- this.attrs.diskStoreName = senderCreation.getDiskStoreName();
- this.attrs.isDiskSynchronous = senderCreation.isDiskSynchronous();
- this.attrs.maximumQueueMemory = senderCreation.getMaximumQueueMemory();
- this.attrs.alertThreshold = senderCreation.getAlertThreshold();
- this.attrs.dispatcherThreads = senderCreation.getDispatcherThreads();
- this.attrs.policy = senderCreation.getOrderPolicy();
+ this.attrs.setParallel(senderCreation.isParallel());
+ this.attrs.setManualStart(senderCreation.isManualStart());
+ this.attrs.setSocketBufferSize(senderCreation.getSocketBufferSize());
+ this.attrs.setSocketReadTimeout(senderCreation.getSocketReadTimeout());
+ this.attrs.setBatchConflationEnabled(senderCreation.isBatchConflationEnabled());
+ this.attrs.setBatchSize(senderCreation.getBatchSize());
+ this.attrs.setBatchTimeInterval(senderCreation.getBatchTimeInterval());
+ this.attrs.setPersistenceEnabled(senderCreation.isPersistenceEnabled());
+ this.attrs.setDiskStoreName(senderCreation.getDiskStoreName());
+ this.attrs.setDiskSynchronous(senderCreation.isDiskSynchronous());
+ this.attrs.setMaximumQueueMemory(senderCreation.getMaximumQueueMemory());
+ this.attrs.setAlertThreshold(senderCreation.getAlertThreshold());
+ this.attrs.setDispatcherThreads(senderCreation.getDispatcherThreads());
+ this.attrs.setOrderPolicy(senderCreation.getOrderPolicy());
for (GatewayEventFilter filter : senderCreation.getGatewayEventFilters()) {
- this.attrs.eventFilters.add(filter);
+ this.attrs.getGatewayEventFilters().add(filter);
}
for (GatewayTransportFilter filter : senderCreation.getGatewayTransportFilters()) {
- this.attrs.transFilters.add(filter);
+ this.attrs.getGatewayTransportFilters().add(filter);
}
- this.attrs.eventSubstitutionFilter = senderCreation.getGatewayEventSubstitutionFilter();
- this.attrs.groupTransactionEvents = senderCreation.mustGroupTransactionEvents();
- this.attrs.enforceThreadsConnectSameReceiver =
- senderCreation.getEnforceThreadsConnectSameReceiver();
+ this.attrs.setEventSubstitutionFilter(senderCreation.getGatewayEventSubstitutionFilter());
+ this.attrs.setGroupTransactionEvents(senderCreation.mustGroupTransactionEvents());
+ this.attrs.setEnforceThreadsConnectSameReceiver(
+ senderCreation.getEnforceThreadsConnectSameReceiver());
}
}
diff --git a/geode-wan/src/test/java/org/apache/geode/cache/wan/internal/parallel/ParallelGatewaySenderImplTest.java b/geode-wan/src/test/java/org/apache/geode/cache/wan/internal/parallel/ParallelGatewaySenderImplTest.java
index 056eac1..84f2261 100644
--- a/geode-wan/src/test/java/org/apache/geode/cache/wan/internal/parallel/ParallelGatewaySenderImplTest.java
+++ b/geode-wan/src/test/java/org/apache/geode/cache/wan/internal/parallel/ParallelGatewaySenderImplTest.java
@@ -53,8 +53,8 @@ public class ParallelGatewaySenderImplTest {
cache = mock(InternalCache.class);
statisticsClock = mock(StatisticsClock.class);
attrs = new GatewaySenderAttributes();
- attrs.isParallel = true;
- attrs.id = "sender";
+ attrs.setParallel(true);
+ attrs.setId("sender");
InternalDistributedSystem system = mock(InternalDistributedSystem.class);
CancelCriterion cancelCriterion = mock(CancelCriterion.class);
Statistics stats = mock(Statistics.class);
@@ -99,7 +99,7 @@ public class ParallelGatewaySenderImplTest {
@Test
public void whenStoppedTwiceCloseInTimeWithGroupTransactionEventsPreStopWaitsTwice() {
- attrs.groupTransactionEvents = true;
+ attrs.setGroupTransactionEvents(true);
gatewaysender = new ParallelGatewaySenderImpl(cache, statisticsClock, attrs);
gatewaysender.start();