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 <[email protected]> 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 <[email protected]> Co-authored-by: Donal Evans <[email protected]> Co-authored-by: Eric Shu <[email protected]> Co-authored-by: Nabarun Nag <[email protected]> --- .../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() {
