You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by je...@apache.org on 2019/10/03 20:52:49 UTC
[geode] 05/11: GEODE-7127: Added gfsh arguments for starting AEQ
with paused event processing.
This is an automated email from the ASF dual-hosted git repository.
jensdeppe pushed a commit to branch release/1.9.2
in repository https://gitbox.apache.org/repos/asf/geode.git
commit a023e198686893b30c4a82a44479cfb6daa1ece5
Author: bross <br...@pivotal.io>
AuthorDate: Tue Aug 27 14:17:26 2019 -0700
GEODE-7127: Added gfsh arguments for starting AEQ with paused event processing.
Co-authored-by: Benjamin Ross <br...@pivotal.io>
Co-authored-by: Donal Evans <do...@pivotal.io>
Co-authored-by: Eric Shu <es...@pivotal.io>
Co-authored-by: Nabarun Nag <na...@cs.wisc.edu>
---
.../CreateAsyncEventQueueCommandDUnitTest.java | 38 +++++++++++++++++++--
.../ListAsyncEventQueuesCommandDUnitTest.java | 26 +++++++++++++++
.../cli/commands/CreateAsyncEventQueueCommand.java | 7 +++-
.../cli/commands/ListAsyncEventQueuesCommand.java | 6 ++--
.../cli/domain/AsyncEventQueueDetails.java | 15 ++++++++-
.../functions/CreateAsyncEventQueueFunction.java | 4 +++
.../functions/ListAsyncEventQueuesFunction.java | 9 ++++-
.../management/internal/cli/i18n/CliStrings.java | 4 +++
.../sanctioned-geode-core-serializables.txt | 2 +-
.../commands/CreateAsyncEventQueueCommandTest.java | 3 ++
.../cli/commands/ListAsyncEventQueuesTest.java | 39 +++++++++++++---------
11 files changed, 130 insertions(+), 23 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandDUnitTest.java
index 13b2523..9054f67 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandDUnitTest.java
@@ -32,7 +32,6 @@ import org.apache.geode.test.junit.categories.AEQTest;
import org.apache.geode.test.junit.rules.GfshCommandRule;
import org.apache.geode.test.junit.rules.MemberStarterRule;
-
@Category({AEQTest.class})
public class CreateAsyncEventQueueCommandDUnitTest {
@@ -70,7 +69,7 @@ public class CreateAsyncEventQueueCommandDUnitTest {
}
@Test
- public void create_sync_event_queue() throws Exception {
+ public void create_async_event_queue() throws Exception {
locator = lsRule.startLocatorVM(0);
lsRule.startServerVM(1, "group1", locator.getPort());
lsRule.startServerVM(2, "group2", locator.getPort());
@@ -110,6 +109,41 @@ public class CreateAsyncEventQueueCommandDUnitTest {
}
@Test
+ public void create_paused_async_event_queue() throws Exception {
+ locator = lsRule.startLocatorVM(0);
+ lsRule.startServerVM(1, locator.getPort());
+ gfsh.connectAndVerify(locator);
+
+ // create queue without start paused set
+ gfsh.executeAndAssertThat(VALID_COMMAND + " --id=queue1 "
+ + "--batch-size=1024 --max-queue-memory=512 --listener-param=param1,param2#value2 ")
+ .statusIsSuccess().tableHasRowCount(1);
+
+
+ // list the queue to verify the the queue has start paused set to false
+ gfsh.executeAndAssertThat("list async-event-queue").statusIsSuccess()
+ .tableHasRowCount(1).tableHasRowWithValues("Member", "ID", "Batch Size",
+ "Persistent", "Disk Store", "Max Memory", "Created with paused event processing",
+ "Currently Paused", "server-1",
+ "queue1", "1024", "false",
+ "null", "512", "false", "false");
+
+ // create queue with start paused set
+ gfsh.executeAndAssertThat(VALID_COMMAND + " --id=queue2 "
+ + "--batch-size=1024 --max-queue-memory=512 --listener-param=param1,param2#value2 --pause-event-processing")
+ .statusIsSuccess().tableHasRowCount(1);
+
+
+ // list the queue to verify the the queue has start paused set to true
+ gfsh.executeAndAssertThat("list async-event-queue").statusIsSuccess()
+ .tableHasRowCount(2).tableHasRowWithValues("Member", "ID", "Batch Size",
+ "Persistent", "Disk Store", "Max Memory", "Created with paused event processing",
+ "Currently Paused", "server-1",
+ "queue2", "1024", "false",
+ "null", "512", "true", "true");
+ }
+
+ @Test
public void create_queue_updates_cc() throws Exception {
locator = lsRule.startLocatorVM(0);
server = lsRule.startServerVM(1, locator.getPort());
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommandDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommandDUnitTest.java
index b877992..040ac95 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommandDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommandDUnitTest.java
@@ -78,5 +78,31 @@ public class ListAsyncEventQueuesCommandDUnitTest {
.tableHasRowWithValues("Member", "ID", "server-1", "queue")
.tableHasRowWithValues("Member", "ID", "server-2", "queue");
+ // Test case where start-paused is set
+ gfsh.executeAndAssertThat("create async-event-queue --id=queue3 --listener="
+ + MyAsyncEventListener.class.getName() + " --pause-event-processing").statusIsSuccess();
+
+ // locator.waitUntilAsyncEventQueuesAreReadyOnExactlyThisManyServers("queue3", 1);
+ gfsh.executeAndAssertThat("list async-event-queue").statusIsSuccess()
+ .tableHasRowCount("Member",6)
+ .tableHasRowWithValues("Member", "ID", "Created with paused event processing",
+ "Currently Paused", "server-1", "queue3",
+ "true", "true")
+ .tableHasRowWithValues("Member", "ID", "Created with paused event processing",
+ "Currently Paused", "server-2", "queue2",
+ "false", "false");
+
+
+ gfsh.executeAndAssertThat("destroy async-event-queue --id=queue").statusIsSuccess();
+ gfsh.executeAndAssertThat("destroy async-event-queue --id=queue1").statusIsSuccess();
+ gfsh.executeAndAssertThat("destroy async-event-queue --id=queue2").statusIsSuccess();
+ gfsh.executeAndAssertThat("destroy async-event-queue --id=queue3").statusIsSuccess();
+ }
+
+
+ @Test
+ public void ensureNoResultIsSuccess() {
+ gfsh.executeAndAssertThat("list async-event-queue").statusIsSuccess()
+ .containsOutput(CliStrings.LIST_ASYNC_EVENT_QUEUES__NO_QUEUES_FOUND_MESSAGE);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommand.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommand.java
index 51a4fae..1273db2 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommand.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommand.java
@@ -87,7 +87,11 @@ public class CreateAsyncEventQueueCommand extends SingleGfshCommand {
@CliOption(key = CliStrings.CREATE_ASYNC_EVENT_QUEUE__LISTENER, mandatory = true,
help = CliStrings.CREATE_ASYNC_EVENT_QUEUE__LISTENER__HELP) String listener,
@CliOption(key = CliStrings.CREATE_ASYNC_EVENT_QUEUE__LISTENER_PARAM_AND_VALUE,
- help = CliStrings.CREATE_ASYNC_EVENT_QUEUE__LISTENER_PARAM_AND_VALUE__HELP) String[] listenerParamsAndValues) {
+ help = CliStrings.CREATE_ASYNC_EVENT_QUEUE__LISTENER_PARAM_AND_VALUE__HELP) String[] listenerParamsAndValues,
+ @CliOption(key = CliStrings.CREATE_ASYNC_EVENT_QUEUE__PAUSE_EVENT_PROCESSING,
+ help = CliStrings.CREATE_ASYNC_EVENT_QUEUE__PAUSE_EVENT_PROCESSING__HELP,
+ unspecifiedDefaultValue = "false",
+ specifiedDefaultValue = "true") boolean pauseEventProcessing) {
if (persistent) {
authorize(ResourcePermission.Resource.CLUSTER, ResourcePermission.Operation.WRITE,
@@ -129,6 +133,7 @@ public class CreateAsyncEventQueueCommand extends SingleGfshCommand {
config.setOrderPolicy(orderPolicy);
config.setParallel(parallel);
config.setPersistent(persistent);
+ config.setPauseEventProcessing(pauseEventProcessing);
CreateAsyncEventQueueFunction function = new CreateAsyncEventQueueFunction();
List<CliFunctionResult> results = executeAndGetFunctionResult(function, config, targetMembers);
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommand.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommand.java
index c414307..919edca 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommand.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesCommand.java
@@ -36,7 +36,8 @@ import org.apache.geode.security.ResourcePermission;
public class ListAsyncEventQueuesCommand extends InternalGfshCommand {
private static final String[] DETAILS_OUTPUT_COLUMNS =
- {"Member", "ID", "Batch Size", "Persistent", "Disk Store", "Max Memory", "Listener"};
+ {"Member", "ID", "Batch Size", "Persistent", "Disk Store", "Max Memory", "Listener",
+ "Created with paused event processing", "Currently Paused"};
private static final String ASYNC_EVENT_QUEUES_TABLE_SECTION = "Async Event Queues";
private static final String MEMBER_ERRORS_TABLE_SECTION = "Member Errors";
@@ -87,7 +88,8 @@ public class ListAsyncEventQueuesCommand extends InternalGfshCommand {
.forEach(entry -> detailsTable.addRow(memberName, entry.getId(),
String.valueOf(entry.getBatchSize()), String.valueOf(entry.isPersistent()),
String.valueOf(entry.getDiskStoreName()), String.valueOf(entry.getMaxQueueMemory()),
- getListenerEntry(entry)));
+ getListenerEntry(entry), String.valueOf(entry.isCreatedWithPausedEventProcessing()),
+ String.valueOf(entry.isPausedEventProcessing())));
});
return result;
}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/domain/AsyncEventQueueDetails.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/domain/AsyncEventQueueDetails.java
index 95f836d..81671cd 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/domain/AsyncEventQueueDetails.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/domain/AsyncEventQueueDetails.java
@@ -32,10 +32,13 @@ public class AsyncEventQueueDetails implements Serializable {
private final int maxQueueMemory;
private final String listener;
private final Properties listenerProperties;
+ private final boolean createdWithPausedEventProcessing;
+ private final boolean pausedEventProcessing;
public AsyncEventQueueDetails(final String id, final int batchSize, final boolean persistent,
final String diskStoreName, final int maxQueueMemory, final String listener,
- final Properties listenerProperties) {
+ final Properties listenerProperties, final boolean createdWithPausedEventProcessing,
+ final boolean pausedEventProcessing) {
this.id = id;
this.batchSize = batchSize;
this.persistent = persistent;
@@ -43,6 +46,8 @@ public class AsyncEventQueueDetails implements Serializable {
this.maxQueueMemory = maxQueueMemory;
this.listener = listener;
this.listenerProperties = listenerProperties;
+ this.createdWithPausedEventProcessing = createdWithPausedEventProcessing;
+ this.pausedEventProcessing = pausedEventProcessing;
}
public String getId() {
@@ -72,4 +77,12 @@ public class AsyncEventQueueDetails implements Serializable {
public Properties getListenerProperties() {
return this.listenerProperties;
}
+
+ public boolean isCreatedWithPausedEventProcessing() {
+ return createdWithPausedEventProcessing;
+ }
+
+ public boolean isPausedEventProcessing() {
+ return pausedEventProcessing;
+ }
}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CreateAsyncEventQueueFunction.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CreateAsyncEventQueueFunction.java
index 1389cb4..6469240 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CreateAsyncEventQueueFunction.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/CreateAsyncEventQueueFunction.java
@@ -83,6 +83,10 @@ public class CreateAsyncEventQueueFunction extends CliFunction {
.setDispatcherThreads(Integer.parseInt(config.getDispatcherThreads()))
.setOrderPolicy(OrderPolicy.valueOf(config.getOrderPolicy()));
+ if (config.isPauseEventProcessing()) {
+ asyncEventQueueFactory.pauseEventDispatching();
+ }
+
String[] gatewayEventFilters = config.getGatewayEventFilters().stream()
.map(ClassNameType::getClassName).toArray(String[]::new);
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/ListAsyncEventQueuesFunction.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/ListAsyncEventQueuesFunction.java
index b0bd9de..25e4517 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/ListAsyncEventQueuesFunction.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/functions/ListAsyncEventQueuesFunction.java
@@ -22,6 +22,7 @@ import java.util.stream.Collectors;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
+import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.internal.cache.xmlcache.Declarable2;
@@ -60,11 +61,17 @@ public class ListAsyncEventQueuesFunction extends CliFunction {
if (listener instanceof Declarable2) {
listenerProperties = ((Declarable2) listener).getConfig();
}
+
return new AsyncEventQueueDetails(queue.getId(), queue.getBatchSize(), queue.isPersistent(),
queue.getDiskStoreName(), queue.getMaximumQueueMemory(), listener.getClass().getName(),
- listenerProperties);
+ listenerProperties, isCreatedWithPausedEventDispatching(queue),
+ queue.isDispatchingPaused());
}).collect(Collectors.toList());
return new CliFunctionResult(memberId, details);
}
+
+ private boolean isCreatedWithPausedEventDispatching(AsyncEventQueue queue) {
+ return ((AsyncEventQueueImpl) queue).getSender().isPaused();
+ }
}
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/i18n/CliStrings.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/i18n/CliStrings.java
index f6b5d1d..9c20491 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/i18n/CliStrings.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/i18n/CliStrings.java
@@ -664,6 +664,10 @@ public class CliStrings {
public static final String CREATE_ASYNC_EVENT_QUEUE__LISTENER_PARAM_AND_VALUE = "listener-param";
public static final String CREATE_ASYNC_EVENT_QUEUE__LISTENER_PARAM_AND_VALUE__HELP =
"Parameter name for the AsyncEventListener. Optionally, parameter names may be followed by # and a value for the parameter. Example: --listener-param=loadAll --listener-param=maxRead#1024";
+ public static final String CREATE_ASYNC_EVENT_QUEUE__PAUSE_EVENT_PROCESSING =
+ "pause-event-processing";
+ public static final String CREATE_ASYNC_EVENT_QUEUE__PAUSE_EVENT_PROCESSING__HELP =
+ "AsyncEventQueue will be created with a paused event processor. resume command must be executed to resume event processing.";
public static final String CREATE_ASYNC_EVENT_QUEUE__GROUP__HELP =
"Group(s) of members on which queue will be created. If no group is specified the queue will be created on all members.";
public static final String CREATE_ASYNC_EVENT_QUEUE__ERROR_WHILE_CREATING_REASON_0 =
diff --git a/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt b/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt
index 586114e..5e6695d 100644
--- a/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt
+++ b/geode-core/src/main/resources/org/apache/geode/internal/sanctioned-geode-core-serializables.txt
@@ -505,7 +505,7 @@ org/apache/geode/management/internal/cli/AbstractCliAroundInterceptor$Response,f
org/apache/geode/management/internal/cli/CliUtil$DeflaterInflaterData,true,1104813333595216795,data:byte[],dataLength:int
org/apache/geode/management/internal/cli/commands/CreateJndiBindingCommand$DATASOURCE_TYPE,false,type:java/lang/String
org/apache/geode/management/internal/cli/commands/ImportClusterConfigurationCommand$Action,false
-org/apache/geode/management/internal/cli/domain/AsyncEventQueueDetails,true,1,batchSize:int,diskStoreName:java/lang/String,id:java/lang/String,listener:java/lang/String,listenerProperties:java/util/Properties,maxQueueMemory:int,persistent:boolean
+org/apache/geode/management/internal/cli/domain/AsyncEventQueueDetails,true,1,batchSize:int,createdWithPausedEventProcessing:boolean,diskStoreName:java/lang/String,id:java/lang/String,listener:java/lang/String,listenerProperties:java/util/Properties,maxQueueMemory:int,pausedEventProcessing:boolean,persistent:boolean
org/apache/geode/management/internal/cli/domain/CacheServerInfo,true,1,bindAddress:java/lang/String,isRunning:boolean,port:int
org/apache/geode/management/internal/cli/domain/ClassName,true,1,className:java/lang/String,initProperties:java/util/Properties
org/apache/geode/management/internal/cli/domain/DataCommandRequest,false,command:java/lang/String,key:java/lang/String,keyClass:java/lang/String,loadOnCacheMiss:boolean,principal:java/lang/Object,putIfAbsent:boolean,query:java/lang/String,recursive:boolean,regionName:java/lang/String,removeAllKeys:java/lang/String,value:java/lang/String,valueClass:java/lang/String
diff --git a/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandTest.java b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandTest.java
index 754c634..d14d3c0 100644
--- a/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/CreateAsyncEventQueueCommandTest.java
@@ -24,6 +24,7 @@ import static org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_AS
import static org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__MAXIMUM_QUEUE_MEMORY;
import static org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__ORDERPOLICY;
import static org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__PARALLEL;
+import static org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__PAUSE_EVENT_PROCESSING;
import static org.apache.geode.management.internal.cli.i18n.CliStrings.CREATE_ASYNC_EVENT_QUEUE__PERSISTENT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
@@ -110,6 +111,8 @@ public class CreateAsyncEventQueueCommandTest {
result = gfsh.parse(COMMAND + "--id=id --listener=xyz --forward-expiration-destroy");
assertThat(result.getParamValue(CREATE_ASYNC_EVENT_QUEUE__FORWARD_EXPIRATION_DESTROY))
.isEqualTo(true);
+ assertThat(result.getParamValue(CREATE_ASYNC_EVENT_QUEUE__PAUSE_EVENT_PROCESSING))
+ .isEqualTo(false);
}
@Test
diff --git a/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesTest.java b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesTest.java
index 56ab979..70f36db 100644
--- a/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/ListAsyncEventQueuesTest.java
@@ -78,7 +78,7 @@ public class ListAsyncEventQueuesTest {
// Mock member's queue details
FakeDetails details = new FakeDetails("server1", "s1-queue-id", 5, true, "diskStoreName", 10,
- "my.listener.class", new Properties());
+ "my.listener.class", new Properties(), false, false);
CliFunctionResult memberResult = new CliFunctionResult(details.getMemberName(),
Collections.singletonList(details.asAsyncEventQueueDetails()));
memberCliResults = Collections.singletonList(memberResult);
@@ -100,7 +100,7 @@ public class ListAsyncEventQueuesTest {
listenerProperties.setProperty("special-property", "special-value");
listenerProperties.setProperty("another-property", "mundane-value");
FakeDetails details = new FakeDetails("server1", "s1-queue-id", 5, true, "diskStoreName", 10,
- "my.listener.class", listenerProperties);
+ "my.listener.class", listenerProperties, false, false);
CliFunctionResult memberResult = new CliFunctionResult(details.getMemberName(),
Collections.singletonList(details.asAsyncEventQueueDetails()));
memberCliResults = Collections.singletonList(memberResult);
@@ -152,21 +152,21 @@ public class ListAsyncEventQueuesTest {
// Mock member's queue details
FakeDetails details1 = new FakeDetails("server1", "s1-queue-id1", 5, false, "diskStoreName", 1,
- "my.listener.class", new Properties());
+ "my.listener.class", new Properties(), false, false);
FakeDetails details2 = new FakeDetails("server1", "s1-queue-id2", 15, true,
- "otherDiskStoreName", 10, "my.listener.class", new Properties());
+ "otherDiskStoreName", 10, "my.listener.class", new Properties(), false, false);
FakeDetails details3 = new FakeDetails("server1", "s1-queue-id3", 25, true, "diskStoreName",
- 100, "my.listener.class", new Properties());
+ 100, "my.listener.class", new Properties(), true, true);
CliFunctionResult member1Result =
new CliFunctionResult("server1", Arrays.asList(details1.asAsyncEventQueueDetails(),
details2.asAsyncEventQueueDetails(), details3.asAsyncEventQueueDetails()));
FakeDetails details4 = new FakeDetails("server2", "s2-queue-id1", 5, false, "diskStoreName", 1,
- "my.listener.class", new Properties());
+ "my.listener.class", new Properties(), false, false);
FakeDetails details5 = new FakeDetails("server2", "s2-queue-id2", 15, true,
- "otherDiskStoreName", 10, "my.listener.class", new Properties());
+ "otherDiskStoreName", 10, "my.listener.class", new Properties(), false, false);
FakeDetails details6 = new FakeDetails("server2", "s2-queue-id3", 25, true, "diskStoreName",
- 100, "my.listener.class", new Properties());
+ 100, "my.listener.class", new Properties(), false, false);
CliFunctionResult member2Result =
new CliFunctionResult("server2", Arrays.asList(details4.asAsyncEventQueueDetails(),
details5.asAsyncEventQueueDetails(), details6.asAsyncEventQueueDetails()));
@@ -192,11 +192,11 @@ public class ListAsyncEventQueuesTest {
// Mock member's queue details
FakeDetails details1 = new FakeDetails("server1", "s1-queue-id1", 5, false, "diskStoreName", 1,
- "my.listener.class", new Properties());
+ "my.listener.class", new Properties(), false, false);
FakeDetails details2 = new FakeDetails("server1", "s1-queue-id2", 15, true,
- "otherDiskStoreName", 10, "my.listener.class", new Properties());
+ "otherDiskStoreName", 10, "my.listener.class", new Properties(), false, false);
FakeDetails details3 = new FakeDetails("server1", "s1-queue-id3", 25, true, "diskStoreName",
- 100, "my.listener.class", new Properties());
+ 100, "my.listener.class", new Properties(), false, false);
CliFunctionResult member1Result =
new CliFunctionResult("server1", Arrays.asList(details1.asAsyncEventQueueDetails(),
details2.asAsyncEventQueueDetails(), details3.asAsyncEventQueueDetails()));
@@ -228,9 +228,12 @@ public class ListAsyncEventQueuesTest {
private int maxQueueMemory;
private String listener;
private Properties listenerProperties;
+ private boolean createWithPausedEventProcessing;
+ private boolean pausedEventProcessing;
private FakeDetails(String memberName, String queueId, int batchSize, boolean persistent,
- String diskStoreName, int maxQueueMemory, String listener, Properties listenerProperties) {
+ String diskStoreName, int maxQueueMemory, String listener, Properties listenerProperties,
+ boolean createWithPausedEventProcessing, boolean pausedEventProcessing) {
this.memberName = memberName;
this.queueId = queueId;
this.batchSize = batchSize;
@@ -239,6 +242,8 @@ public class ListAsyncEventQueuesTest {
this.maxQueueMemory = maxQueueMemory;
this.listener = listener;
this.listenerProperties = listenerProperties;
+ this.createWithPausedEventProcessing = createWithPausedEventProcessing;
+ this.pausedEventProcessing = pausedEventProcessing;
}
public String getMemberName() {
@@ -247,13 +252,17 @@ public class ListAsyncEventQueuesTest {
private AsyncEventQueueDetails asAsyncEventQueueDetails() {
return new AsyncEventQueueDetails(queueId, batchSize, persistent, diskStoreName,
- maxQueueMemory, listener, listenerProperties);
+ maxQueueMemory, listener, listenerProperties, createWithPausedEventProcessing,
+ pausedEventProcessing);
}
private String[] expectedRowHeaderAndValue() {
return new String[] {"Member", "ID", "Batch Size", "Persistent", "Disk Store", "Max Memory",
- "Listener", memberName, queueId, String.valueOf(batchSize), String.valueOf(persistent),
- diskStoreName, String.valueOf(maxQueueMemory), expectedListenerOutput()};
+ "Listener", "Created with paused event processing", "Currently Paused", memberName,
+ queueId, String.valueOf(batchSize),
+ String.valueOf(persistent),
+ diskStoreName, String.valueOf(maxQueueMemory), expectedListenerOutput(),
+ String.valueOf(createWithPausedEventProcessing), String.valueOf(pausedEventProcessing)};
}
private String expectedListenerOutput() {