[
https://issues.apache.org/jira/browse/GEODE-8079?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17102386#comment-17102386
]
ASF GitHub Bot commented on GEODE-8079:
---------------------------------------
jujoramos commented on a change in pull request #5070:
URL: https://github.com/apache/geode/pull/5070#discussion_r422027492
##########
File path:
geode-core/src/test/java/org/apache/geode/internal/cache/DistributedRegionTest.java
##########
@@ -180,4 +189,75 @@ public void
regionSyncNotInvokedInPerformSynchronizeForLostMemberTaskIfRegionNot
verify(distributedRegion, never()).synchronizeForLostMember(member,
lostMemberVersionID);
}
+
+ @Test
+ public void
validateAsynchronousEventDispatcherShouldDoNothingWhenDispatcherIdCanNotBeFound()
{
+ InternalCache internalCache = mock(InternalCache.class);
+ when(internalCache.getAllGatewaySenders())
+ .thenReturn(Collections.singleton(mock(GatewaySender.class)));
+ DistributedRegion distributedRegion = mock(DistributedRegion.class);
+ when(distributedRegion.getCache()).thenReturn(internalCache);
+
doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString());
+
+
distributedRegion.validateAsynchronousEventDispatcher("nonExistingDispatcher");
+ }
+
+ @Test
+ public void
validateAsynchronousEventDispatcherShouldDoNothingWhenFoundDispatcherIsSerial()
{
+ String senderId = "mySender";
+ GatewaySender serialSender = mock(GatewaySender.class);
+ when(serialSender.isParallel()).thenReturn(false);
+ when(serialSender.getId()).thenReturn(senderId);
+ InternalCache internalCache = mock(InternalCache.class);
+
when(internalCache.getAllGatewaySenders()).thenReturn(Collections.singleton(serialSender));
+ DistributedRegion distributedRegion = mock(DistributedRegion.class);
+ when(distributedRegion.getCache()).thenReturn(internalCache);
+
doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString());
+
+ distributedRegion.validateAsynchronousEventDispatcher(senderId);
+ }
+
+ @Test
+ public void
validateAsynchronousEventDispatcherShouldThrowExceptionWhenDispatcherIdMatchesAnExistingParallelAsyncEventQueue()
{
+ String senderId = "senderId";
+ String regionPath = "thisRegion";
+ String internalSenderId = getSenderIdFromAsyncEventQueueId(senderId);
+ GatewaySender parallelAsyncEventQueue = mock(GatewaySender.class);
+ when(parallelAsyncEventQueue.isParallel()).thenReturn(true);
+ when(parallelAsyncEventQueue.getId()).thenReturn(internalSenderId);
+ InternalCache internalCache = mock(InternalCache.class);
+ when(internalCache.getAllGatewaySenders())
+ .thenReturn(Collections.singleton(parallelAsyncEventQueue));
+ DistributedRegion distributedRegion = mock(DistributedRegion.class);
+ when(distributedRegion.getCache()).thenReturn(internalCache);
+ when(distributedRegion.getFullPath()).thenReturn(regionPath);
+
doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString());
+
+ assertThatThrownBy(
+ () ->
distributedRegion.validateAsynchronousEventDispatcher(internalSenderId))
+ .isInstanceOf(AsyncEventQueueConfigurationException.class)
+ .hasMessage("Parallel Async Event Queue " + senderId
+ + " can not be used with replicated region " + regionPath);
+ }
+
+ @Test
+ public void
validateAsynchronousEventDispatcherShouldThrowExceptionWhenDispatcherIdMatchesAnExistingParallelGatewaySender()
{
+ String senderId = "senderId";
+ String regionPath = "thisRegion";
+ GatewaySender parallelGatewaySender = mock(GatewaySender.class);
+ when(parallelGatewaySender.isParallel()).thenReturn(true);
+ when(parallelGatewaySender.getId()).thenReturn(senderId);
+ InternalCache internalCache = mock(InternalCache.class);
+ when(internalCache.getAllGatewaySenders())
+ .thenReturn(Collections.singleton(parallelGatewaySender));
+ DistributedRegion distributedRegion = mock(DistributedRegion.class);
Review comment:
See my comments above.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/AlterRegionCommandDUnitTest.java
##########
@@ -343,4 +346,76 @@ public void
alterPartitionPersistentRegionWithParallelNonPersistentGatewaySender
assertThat(regionConfig.getRegionAttributes().getGatewaySenderIds()).isBlank();
});
}
+
+ @Test
+ public void
alterReplicateRegionWithParallelGatewaySenderShouldFailAndDoNotPersistChangesIntoTheClusterConfigurationService()
{
+ addIgnoredException(GatewaySenderConfigurationException.class);
+ addIgnoredException("could not get remote locator information");
+ String regionName = testName.getMethodName();
+ String gatewaySenderId = testName.getMethodName() +
"_parallelGatewaySender";
+
+ gfsh.executeAndAssertThat(
+ "create gateway-sender --parallel=true --enable-persistence=false
--remote-distributed-system-id=2 --id="
Review comment:
Done!.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/AlterRegionCommandDUnitTest.java
##########
@@ -343,4 +346,76 @@ public void
alterPartitionPersistentRegionWithParallelNonPersistentGatewaySender
assertThat(regionConfig.getRegionAttributes().getGatewaySenderIds()).isBlank();
});
}
+
+ @Test
+ public void
alterReplicateRegionWithParallelGatewaySenderShouldFailAndDoNotPersistChangesIntoTheClusterConfigurationService()
{
+ addIgnoredException(GatewaySenderConfigurationException.class);
+ addIgnoredException("could not get remote locator information");
+ String regionName = testName.getMethodName();
+ String gatewaySenderId = testName.getMethodName() +
"_parallelGatewaySender";
+
+ gfsh.executeAndAssertThat(
+ "create gateway-sender --parallel=true --enable-persistence=false
--remote-distributed-system-id=2 --id="
+ + gatewaySenderId)
+ .statusIsSuccess().doesNotContainOutput("Did not complete waiting");
+ gfsh.executeAndAssertThat("create region --type=REPLICATE --name=" +
regionName)
+ .statusIsSuccess();
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR +
regionName, 1);
+
+ // Associate the gateway-sender
+ gfsh.executeAndAssertThat(
+ "alter region --name=" + regionName + " --gateway-sender-id=" +
gatewaySenderId)
Review comment:
Done!.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/AlterRegionCommandDUnitTest.java
##########
@@ -343,4 +346,76 @@ public void
alterPartitionPersistentRegionWithParallelNonPersistentGatewaySender
assertThat(regionConfig.getRegionAttributes().getGatewaySenderIds()).isBlank();
});
}
+
+ @Test
+ public void
alterReplicateRegionWithParallelGatewaySenderShouldFailAndDoNotPersistChangesIntoTheClusterConfigurationService()
{
+ addIgnoredException(GatewaySenderConfigurationException.class);
+ addIgnoredException("could not get remote locator information");
+ String regionName = testName.getMethodName();
+ String gatewaySenderId = testName.getMethodName() +
"_parallelGatewaySender";
+
+ gfsh.executeAndAssertThat(
+ "create gateway-sender --parallel=true --enable-persistence=false
--remote-distributed-system-id=2 --id="
+ + gatewaySenderId)
+ .statusIsSuccess().doesNotContainOutput("Did not complete waiting");
+ gfsh.executeAndAssertThat("create region --type=REPLICATE --name=" +
regionName)
+ .statusIsSuccess();
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR +
regionName, 1);
+
+ // Associate the gateway-sender
+ gfsh.executeAndAssertThat(
+ "alter region --name=" + regionName + " --gateway-sender-id=" +
gatewaySenderId)
+ .statusIsError().containsOutput("server-1", "Parallel Gateway Sender "
+ gatewaySenderId
+ + " can not be used with replicated region " + SEPARATOR +
regionName);
+
+ // Check the cluster configuration service.
+ locator.invoke(() -> {
+ InternalLocator internalLocator = ClusterStartupRule.getLocator();
+ assertThat(internalLocator).isNotNull();
+ CacheConfig config =
+
internalLocator.getConfigurationPersistenceService().getCacheConfig("cluster");
+
+ RegionConfig regionConfig = find(config.getRegions(), regionName);
+ assertThat(regionConfig).isNotNull();
+ assertThat(regionConfig.getRegionAttributes()).isNotNull();
+
assertThat(regionConfig.getRegionAttributes().getGatewaySenderIds()).isNull();
+ });
+ }
+
+ @Test
+ public void
alterReplicateRegionWithParallelAsynchronousEventQueueShouldFailAndDoNotPersistChangesIntoTheClusterConfigurationService()
{
+ addIgnoredException(AsyncEventQueueConfigurationException.class);
+ String regionName = testName.getMethodName();
+ String asyncEventQueueName = testName.getMethodName() + "_asyncEventQueue";
+
+ gfsh.executeAndAssertThat(
+ "create async-event-queue --parallel=true --persistent=false
--listener=org.apache.geode.internal.cache.wan.MyAsyncEventListener --id="
Review comment:
Done!.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/management/internal/cli/commands/AlterRegionCommandDUnitTest.java
##########
@@ -343,4 +346,76 @@ public void
alterPartitionPersistentRegionWithParallelNonPersistentGatewaySender
assertThat(regionConfig.getRegionAttributes().getGatewaySenderIds()).isBlank();
});
}
+
+ @Test
+ public void
alterReplicateRegionWithParallelGatewaySenderShouldFailAndDoNotPersistChangesIntoTheClusterConfigurationService()
{
+ addIgnoredException(GatewaySenderConfigurationException.class);
+ addIgnoredException("could not get remote locator information");
+ String regionName = testName.getMethodName();
+ String gatewaySenderId = testName.getMethodName() +
"_parallelGatewaySender";
+
+ gfsh.executeAndAssertThat(
+ "create gateway-sender --parallel=true --enable-persistence=false
--remote-distributed-system-id=2 --id="
+ + gatewaySenderId)
+ .statusIsSuccess().doesNotContainOutput("Did not complete waiting");
+ gfsh.executeAndAssertThat("create region --type=REPLICATE --name=" +
regionName)
+ .statusIsSuccess();
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR +
regionName, 1);
+
+ // Associate the gateway-sender
+ gfsh.executeAndAssertThat(
+ "alter region --name=" + regionName + " --gateway-sender-id=" +
gatewaySenderId)
+ .statusIsError().containsOutput("server-1", "Parallel Gateway Sender "
+ gatewaySenderId
+ + " can not be used with replicated region " + SEPARATOR +
regionName);
+
+ // Check the cluster configuration service.
+ locator.invoke(() -> {
+ InternalLocator internalLocator = ClusterStartupRule.getLocator();
+ assertThat(internalLocator).isNotNull();
+ CacheConfig config =
+
internalLocator.getConfigurationPersistenceService().getCacheConfig("cluster");
+
+ RegionConfig regionConfig = find(config.getRegions(), regionName);
+ assertThat(regionConfig).isNotNull();
+ assertThat(regionConfig.getRegionAttributes()).isNotNull();
+
assertThat(regionConfig.getRegionAttributes().getGatewaySenderIds()).isNull();
+ });
+ }
+
+ @Test
+ public void
alterReplicateRegionWithParallelAsynchronousEventQueueShouldFailAndDoNotPersistChangesIntoTheClusterConfigurationService()
{
+ addIgnoredException(AsyncEventQueueConfigurationException.class);
+ String regionName = testName.getMethodName();
+ String asyncEventQueueName = testName.getMethodName() + "_asyncEventQueue";
+
+ gfsh.executeAndAssertThat(
+ "create async-event-queue --parallel=true --persistent=false
--listener=org.apache.geode.internal.cache.wan.MyAsyncEventListener --id="
+ + asyncEventQueueName)
+ .statusIsSuccess();
+
locator.waitUntilAsyncEventQueuesAreReadyOnExactlyThisManyServers(asyncEventQueueName,
1);
+
+ gfsh.executeAndAssertThat("create region --type=REPLICATE --name=" +
regionName)
+ .statusIsSuccess();
+ locator.waitUntilRegionIsReadyOnExactlyThisManyServers(SEPARATOR +
regionName, 1);
+
+ // Associate the async-event-queue
+ gfsh.executeAndAssertThat(
+ "alter region --name=" + regionName + " --async-event-queue-id=" +
asyncEventQueueName)
Review comment:
Done!.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> AttributesMutator Should Validate AsyncEventQueue/GatewaySender Type
> --------------------------------------------------------------------
>
> Key: GEODE-8079
> URL: https://issues.apache.org/jira/browse/GEODE-8079
> Project: Geode
> Issue Type: Bug
> Components: configuration, gfsh, wan
> Affects Versions: 1.12.0
> Reporter: Juan Ramos
> Assignee: Juan Ramos
> Priority: Major
> Labels: caching-applications
>
> By design, a parallel {{gateway-sender}} can't be attached to a {{REPLICATE}}
> region.
> While working on on GEODE-8029 I've found that the above fact is correctly
> validated when creating or initialising the region, but totally ignored when
> updating the region through the {{AttributesMutator}} class.
> Altering a {{REPLICATE}} region to dispatch events through a parallel
> {{gateway-sender}} results in cryptic errors while putting entries into the
> region afterwards:
> {noformat}
> [vm1] [warn 2020/05/06 10:34:09.638 IST <RMI TCP Connection(1)-127.0.0.1>
> tid=0x13] GatewaySender: Not queuing the event
> GatewaySenderEventImpl[id=EventID[id=18
> bytes;threadID=0x10062|2;sequenceID=91;bucketId=98];action=0;operation=CREATE;region=/TestRegion;key=Key90;value=Value90;valueIsObject=1;numberOfParts=9;callbackArgument=GatewaySenderEventCallbackArgument
>
> [originalCallbackArg=null;originatingSenderId=1;recipientGatewayReceivers={2}];possibleDuplicate=false;creationTime=1588757649638;shadowKey=-1;timeStamp=1588757649638;acked=false;dispatched=false;bucketId=98;isConcurrencyConflict=false],
> as the region for which this event originated is not yet configured in the
> GatewaySender
> [vm1] [warn 2020/05/06 10:34:09.638 IST <RMI TCP Connection(1)-127.0.0.1>
> tid=0x13] GatewaySender: Not queuing the event
> GatewaySenderEventImpl[id=EventID[id=18
> bytes;threadID=0x10063|2;sequenceID=92;bucketId=99];action=0;operation=CREATE;region=/TestRegion;key=Key91;value=Value91;valueIsObject=1;numberOfParts=9;callbackArgument=GatewaySenderEventCallbackArgument
>
> [originalCallbackArg=null;originatingSenderId=1;recipientGatewayReceivers={2}];possibleDuplicate=false;creationTime=1588757649638;shadowKey=-1;timeStamp=1588757649638;acked=false;dispatched=false;bucketId=99;isConcurrencyConflict=false],
> as the region for which this event originated is not yet configured in the
> GatewaySender
> [vm1] [warn 2020/05/06 10:34:09.639 IST <RMI TCP Connection(1)-127.0.0.1>
> tid=0x13] GatewaySender: Not queuing the event
> GatewaySenderEventImpl[id=EventID[id=18
> bytes;threadID=0x10064|2;sequenceID=93;bucketId=100];action=0;operation=CREATE;region=/TestRegion;key=Key92;value=Value92;valueIsObject=1;numberOfParts=9;callbackArgument=GatewaySenderEventCallbackArgument
>
> [originalCallbackArg=null;originatingSenderId=1;recipientGatewayReceivers={2}];possibleDuplicate=false;creationTime=1588757649638;shadowKey=-1;timeStamp=1588757649638;acked=false;dispatched=false;bucketId=100;isConcurrencyConflict=false],
> as the region for which this event originated is not yet configured in the
> GatewaySender
> [vm1] [warn 2020/05/06 10:34:09.639 IST <RMI TCP Connection(1)-127.0.0.1>
> tid=0x13] GatewaySender: Not queuing the event
> GatewaySenderEventImpl[id=EventID[id=18
> bytes;threadID=0x10065|2;sequenceID=94;bucketId=101];action=0;operation=CREATE;region=/TestRegion;key=Key93;value=Value93;valueIsObject=1;numberOfParts=9;callbackArgument=GatewaySenderEventCallbackArgument
>
> [originalCallbackArg=null;originatingSenderId=1;recipientGatewayReceivers={2}];possibleDuplicate=false;creationTime=1588757649639;shadowKey=-1;timeStamp=1588757649639;acked=false;dispatched=false;bucketId=101;isConcurrencyConflict=false],
> as the region for which this event originated is not yet configured in the
> GatewaySender
> {noformat}
> When done from {{GFSH}}, on the other hand, the server doesn't even start up
> after altering the region as the {{cluster-configuration}} is invalid:
> {noformat}
> gfsh -e "connect" -e "create region --name=TestRegion --type=REPLICATE"
> Member | Status | Message
> --------------- | ------ | -------------------------------------------------
> cluster1-server | OK | Region "/TestRegion" created on "cluster1-server"
> Cluster configuration for group 'cluster' is updated.
> gfsh -e "connect" -e "create gateway-sender --id=MyGateway
> --remote-distributed-system-id=2 --parallel=true"
> Member | Status | Message
> --------------- | ------ |
> ------------------------------------------------------
> cluster1-server | OK | GatewaySender "MyGateway" created on
> "cluster1-server"
> Cluster configuration for group 'cluster' is updated.
> gfsh -e "connect" -e "alter region --name=/TestRegion
> -–gateway-sender-id=MyGateway"
> Member | Status | Message
> --------------- | ------ | -------------------------
> cluster1-server | OK | Region TestRegion altered
> Cluster configuration for group 'cluster' is updated.
> ////// Restart Cluster
> [warn 2020/05/06 10:09:07.385 IST <main> tid=0x1] Initialization failed for
> Region /TestRegion
> org.apache.geode.internal.cache.wan.GatewaySenderConfigurationException:
> Parallel gateway sender MyGateway can not be used with replicated region
> /TestRegion
> at
> org.apache.geode.internal.cache.DistributedRegion.preInitialize(DistributedRegion.java:1035)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.createVMRegion(GemFireCacheImpl.java:2996)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.basicCreateRegion(GemFireCacheImpl.java:2891)
> at
> org.apache.geode.internal.cache.xmlcache.RegionCreation.createRoot(RegionCreation.java:237)
> at
> org.apache.geode.internal.cache.xmlcache.CacheCreation.initializeRegions(CacheCreation.java:658)
> at
> org.apache.geode.internal.cache.xmlcache.CacheCreation.create(CacheCreation.java:592)
> at
> org.apache.geode.internal.cache.xmlcache.CacheXmlParser.create(CacheXmlParser.java:339)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.loadCacheXml(GemFireCacheImpl.java:4096)
> at
> org.apache.geode.internal.cache.ClusterConfigurationLoader.applyClusterXmlConfiguration(ClusterConfigurationLoader.java:208)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.applyJarAndXmlFromClusterConfig(GemFireCacheImpl.java:1409)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.initialize(GemFireCacheImpl.java:1374)
> at
> org.apache.geode.internal.cache.InternalCacheBuilder.create(InternalCacheBuilder.java:191)
> at
> org.apache.geode.internal.cache.InternalCacheBuilder.create(InternalCacheBuilder.java:158)
> at org.apache.geode.cache.CacheFactory.create(CacheFactory.java:142)
> at
> org.apache.geode.distributed.internal.DefaultServerLauncherCacheProvider.createCache(DefaultServerLauncherCacheProvider.java:52)
> at
> org.apache.geode.distributed.ServerLauncher.createCache(ServerLauncher.java:892)
> at
> org.apache.geode.distributed.ServerLauncher.start(ServerLauncher.java:807)
> at
> org.apache.geode.distributed.ServerLauncher.run(ServerLauncher.java:737)
> at
> org.apache.geode.distributed.ServerLauncher.main(ServerLauncher.java:256)
> {noformat}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
