[
https://issues.apache.org/jira/browse/GEODE-8029?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17098502#comment-17098502
]
ASF GitHub Bot commented on GEODE-8029:
---------------------------------------
jujoramos commented on a change in pull request #5037:
URL: https://github.com/apache/geode/pull/5037#discussion_r419131931
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/PersistentGatewaySenderWithIsolatedDiskStoreDistributedTest.java
##########
@@ -0,0 +1,346 @@
+/*
+ * 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;
+
+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.getRandomAvailableTCPPortsForDUnitSite;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.geode.cache.DiskStore;
+import org.apache.geode.cache.DiskStoreFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.wan.GatewayReceiverFactory;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.cache.wan.GatewaySenderFactory;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.cache.DiskStoreImpl;
+import org.apache.geode.internal.cache.RegionQueue;
+import org.apache.geode.internal.cache.persistence.OplogType;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.DistributedDiskDirRule;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
+
+/**
+ * Tests to verify WAN functionality when the gateway-sender(s) have isolated,
non-shared with
+ * other region(s), disk-store(s).
+ */
+@RunWith(JUnitParamsRunner.class)
+public class PersistentGatewaySenderWithIsolatedDiskStoreDistributedTest
implements Serializable {
+ private static final String REGION_NAME = "TestRegion";
+ private static final String DISK_STORE_ID = "testDisk";
+ private static final String GATEWAY_SENDER_ID = "testSender";
+ private static final String TEST_CASE_NAME = "[{index}]
{method}(RegionType:{0}, Parallel:{1})";
+ private int site1Port, site2Port;
+ private VM serverCluster1, serverCluster2;
+
+ @Rule
+ public CacheRule cacheRule = new CacheRule();
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule();
+
+ @Rule
+ public SerializableTestName testName = new SerializableTestName();
+
+ @Rule
+ public DistributedDiskDirRule distributedDiskDirRule = new
DistributedDiskDirRule();
+
+ private Properties createLocatorConfiguration(int distributedSystemId, int
localLocatorPort,
+ int remoteLocatorPort) {
+ Properties config = new Properties();
+ config.setProperty(MCAST_PORT, "0");
+ config.setProperty(DISTRIBUTED_SYSTEM_ID,
String.valueOf(distributedSystemId));
+ config.setProperty(LOCATORS, "localhost[" + localLocatorPort + ']');
+ config.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocatorPort +
']');
+ config.setProperty(START_LOCATOR,
+ "localhost[" + localLocatorPort +
"],server=true,peer=true,hostname-for-clients=localhost");
+
+ return config;
+ }
+
+ private Properties createServerConfiguration(int localLocatorPort) {
+ Properties config = new Properties();
+ config.setProperty(MCAST_PORT, "0");
+ config.setProperty(LOCATORS, "localhost[" + localLocatorPort + ']');
+
+ return config;
+ }
+
+ private void createDiskStore() {
+ String basePath = distributedDiskDirRule.getDiskDir().getAbsolutePath();
+ File diskDirectory = new File(basePath + File.separator + DISK_STORE_ID);
+ DiskStoreFactory diskStoreFactory =
cacheRule.getCache().createDiskStoreFactory();
+ diskStoreFactory.setAutoCompact(true);
+ diskStoreFactory.setAllowForceCompaction(true);
+ diskStoreFactory.setDiskDirs(new File[] {diskDirectory});
+ diskStoreFactory.create(DISK_STORE_ID);
+ }
+
+ private void createRegion(RegionShortcut regionShortcut) {
+ cacheRule.getCache()
+ .<String, String>createRegionFactory(regionShortcut)
+ .create(REGION_NAME);
+ }
+
+ private void createGatewayReceiver() {
+ GatewayReceiverFactory gatewayReceiverFactory =
+ cacheRule.getCache().createGatewayReceiverFactory();
+ gatewayReceiverFactory.setManualStart(false);
+ gatewayReceiverFactory.create();
+ }
+
+ private void createGatewaySender(boolean parallel, int
remoteDistributedSystemId) {
+ GatewaySenderFactory gatewaySenderFactory =
cacheRule.getCache().createGatewaySenderFactory();
+ gatewaySenderFactory.setParallel(parallel);
+ gatewaySenderFactory.setDiskSynchronous(true);
+ gatewaySenderFactory.setPersistenceEnabled(true);
+ gatewaySenderFactory.setDiskStoreName(DISK_STORE_ID);
+ gatewaySenderFactory.create(GATEWAY_SENDER_ID, remoteDistributedSystemId);
+ }
+
+ private void createServerWithRegionAndGatewayReceiver(RegionShortcut
regionShortcut) {
+ createGatewayReceiver();
+ createRegion(regionShortcut);
+ }
+
+ private void createServerWithRegionAndPersistentGatewaySender(RegionShortcut
regionShortcut,
+ int remoteDistributedSystemId, boolean parallel) {
+ createDiskStore();
+ createRegion(regionShortcut);
+ createGatewaySender(parallel, remoteDistributedSystemId);
+ Region<String, String> region =
cacheRule.getCache().getRegion(REGION_NAME);
+ region.getAttributesMutator().addGatewaySenderId(GATEWAY_SENDER_ID);
+ }
+
+ private void gracefullyDisconnect() {
+
InternalDistributedSystem.getConnectedInstance().stopReconnectingNoDisconnect();
+ InternalDistributedSystem.getConnectedInstance().disconnect();
+ await()
+ .untilAsserted(() ->
assertThat(InternalDistributedSystem.getConnectedInstance()).isNull());
+ }
+
+ private void awaitForQueueSize(int queueSize) {
+ GatewaySender gatewaySender =
cacheRule.getCache().getGatewaySender(GATEWAY_SENDER_ID);
+ await().untilAsserted(() -> {
+ Set<RegionQueue> queues = ((AbstractGatewaySender)
gatewaySender).getQueues();
+ int totalSize = queues.stream().mapToInt(RegionQueue::size).sum();
+ assertThat(queueSize).isEqualTo(totalSize);
+ });
+ }
+
+ @SuppressWarnings("unused")
+ static Object[] regionAndGatewayTypes() {
+ ArrayList<Object[]> parameters = new ArrayList<>();
+ parameters.add(new Object[] {RegionShortcut.PARTITION, true});
+ parameters.add(new Object[] {RegionShortcut.REPLICATE, false});
Review comment:
It doesn't make much sense to use a `PARTITION` region with a serial
`gateway-sender`, and the other combination is not even supported (`REPLICATE`
region with parallel `gateway-sender`). I've added the extra supported
combination, anyways, it's harmless, and it's always good to have extra tests.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/PersistentGatewaySenderWithIsolatedDiskStoreDistributedTest.java
##########
@@ -0,0 +1,346 @@
+/*
+ * 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;
+
+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.getRandomAvailableTCPPortsForDUnitSite;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.geode.cache.DiskStore;
+import org.apache.geode.cache.DiskStoreFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.wan.GatewayReceiverFactory;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.cache.wan.GatewaySenderFactory;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.cache.DiskStoreImpl;
+import org.apache.geode.internal.cache.RegionQueue;
+import org.apache.geode.internal.cache.persistence.OplogType;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.DistributedDiskDirRule;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
+
+/**
+ * Tests to verify WAN functionality when the gateway-sender(s) have isolated,
non-shared with
+ * other region(s), disk-store(s).
+ */
+@RunWith(JUnitParamsRunner.class)
+public class PersistentGatewaySenderWithIsolatedDiskStoreDistributedTest
implements Serializable {
+ private static final String REGION_NAME = "TestRegion";
+ private static final String DISK_STORE_ID = "testDisk";
+ private static final String GATEWAY_SENDER_ID = "testSender";
+ private static final String TEST_CASE_NAME = "[{index}]
{method}(RegionType:{0}, Parallel:{1})";
+ private int site1Port, site2Port;
+ private VM serverCluster1, serverCluster2;
+
+ @Rule
+ public CacheRule cacheRule = new CacheRule();
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule();
+
+ @Rule
+ public SerializableTestName testName = new SerializableTestName();
+
+ @Rule
+ public DistributedDiskDirRule distributedDiskDirRule = new
DistributedDiskDirRule();
+
+ private Properties createLocatorConfiguration(int distributedSystemId, int
localLocatorPort,
+ int remoteLocatorPort) {
+ Properties config = new Properties();
+ config.setProperty(MCAST_PORT, "0");
+ config.setProperty(DISTRIBUTED_SYSTEM_ID,
String.valueOf(distributedSystemId));
+ config.setProperty(LOCATORS, "localhost[" + localLocatorPort + ']');
+ config.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocatorPort +
']');
+ config.setProperty(START_LOCATOR,
+ "localhost[" + localLocatorPort +
"],server=true,peer=true,hostname-for-clients=localhost");
+
+ return config;
+ }
+
+ private Properties createServerConfiguration(int localLocatorPort) {
+ Properties config = new Properties();
+ config.setProperty(MCAST_PORT, "0");
+ config.setProperty(LOCATORS, "localhost[" + localLocatorPort + ']');
+
+ return config;
+ }
+
+ private void createDiskStore() {
+ String basePath = distributedDiskDirRule.getDiskDir().getAbsolutePath();
+ File diskDirectory = new File(basePath + File.separator + DISK_STORE_ID);
+ DiskStoreFactory diskStoreFactory =
cacheRule.getCache().createDiskStoreFactory();
+ diskStoreFactory.setAutoCompact(true);
+ diskStoreFactory.setAllowForceCompaction(true);
+ diskStoreFactory.setDiskDirs(new File[] {diskDirectory});
+ diskStoreFactory.create(DISK_STORE_ID);
+ }
+
+ private void createRegion(RegionShortcut regionShortcut) {
+ cacheRule.getCache()
+ .<String, String>createRegionFactory(regionShortcut)
+ .create(REGION_NAME);
+ }
+
+ private void createGatewayReceiver() {
+ GatewayReceiverFactory gatewayReceiverFactory =
+ cacheRule.getCache().createGatewayReceiverFactory();
+ gatewayReceiverFactory.setManualStart(false);
+ gatewayReceiverFactory.create();
+ }
+
+ private void createGatewaySender(boolean parallel, int
remoteDistributedSystemId) {
+ GatewaySenderFactory gatewaySenderFactory =
cacheRule.getCache().createGatewaySenderFactory();
+ gatewaySenderFactory.setParallel(parallel);
+ gatewaySenderFactory.setDiskSynchronous(true);
+ gatewaySenderFactory.setPersistenceEnabled(true);
+ gatewaySenderFactory.setDiskStoreName(DISK_STORE_ID);
+ gatewaySenderFactory.create(GATEWAY_SENDER_ID, remoteDistributedSystemId);
+ }
+
+ private void createServerWithRegionAndGatewayReceiver(RegionShortcut
regionShortcut) {
+ createGatewayReceiver();
+ createRegion(regionShortcut);
+ }
+
+ private void createServerWithRegionAndPersistentGatewaySender(RegionShortcut
regionShortcut,
+ int remoteDistributedSystemId, boolean parallel) {
+ createDiskStore();
+ createRegion(regionShortcut);
+ createGatewaySender(parallel, remoteDistributedSystemId);
+ Region<String, String> region =
cacheRule.getCache().getRegion(REGION_NAME);
+ region.getAttributesMutator().addGatewaySenderId(GATEWAY_SENDER_ID);
+ }
+
+ private void gracefullyDisconnect() {
+
InternalDistributedSystem.getConnectedInstance().stopReconnectingNoDisconnect();
+ InternalDistributedSystem.getConnectedInstance().disconnect();
+ await()
+ .untilAsserted(() ->
assertThat(InternalDistributedSystem.getConnectedInstance()).isNull());
+ }
+
+ private void awaitForQueueSize(int queueSize) {
+ GatewaySender gatewaySender =
cacheRule.getCache().getGatewaySender(GATEWAY_SENDER_ID);
+ await().untilAsserted(() -> {
+ Set<RegionQueue> queues = ((AbstractGatewaySender)
gatewaySender).getQueues();
+ int totalSize = queues.stream().mapToInt(RegionQueue::size).sum();
+ assertThat(queueSize).isEqualTo(totalSize);
+ });
+ }
+
+ @SuppressWarnings("unused")
+ static Object[] regionAndGatewayTypes() {
+ ArrayList<Object[]> parameters = new ArrayList<>();
+ parameters.add(new Object[] {RegionShortcut.PARTITION, true});
+ parameters.add(new Object[] {RegionShortcut.REPLICATE, false});
+
+ return parameters.toArray();
+ }
+
+ @Before
+ public void setUp() {
+ VM locatorCluster1 = getVM(0);
+ serverCluster1 = getVM(1);
+ VM locatorCluster2 = getVM(2);
+ serverCluster2 = getVM(3);
Review comment:
It's irrelevant for this scenario, we're interested only in the
`disk-stores` locally used by the servers.
##########
File path:
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/PersistentGatewaySenderWithIsolatedDiskStoreDistributedTest.java
##########
@@ -0,0 +1,346 @@
+/*
+ * 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;
+
+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.getRandomAvailableTCPPortsForDUnitSite;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.geode.cache.DiskStore;
+import org.apache.geode.cache.DiskStoreFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.wan.GatewayReceiverFactory;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.cache.wan.GatewaySenderFactory;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.cache.DiskStoreImpl;
+import org.apache.geode.internal.cache.RegionQueue;
+import org.apache.geode.internal.cache.persistence.OplogType;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.DistributedDiskDirRule;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
+
+/**
+ * Tests to verify WAN functionality when the gateway-sender(s) have isolated,
non-shared with
+ * other region(s), disk-store(s).
+ */
+@RunWith(JUnitParamsRunner.class)
+public class PersistentGatewaySenderWithIsolatedDiskStoreDistributedTest
implements Serializable {
+ private static final String REGION_NAME = "TestRegion";
+ private static final String DISK_STORE_ID = "testDisk";
+ private static final String GATEWAY_SENDER_ID = "testSender";
+ private static final String TEST_CASE_NAME = "[{index}]
{method}(RegionType:{0}, Parallel:{1})";
+ private int site1Port, site2Port;
+ private VM serverCluster1, serverCluster2;
+
+ @Rule
+ public CacheRule cacheRule = new CacheRule();
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule();
+
+ @Rule
+ public SerializableTestName testName = new SerializableTestName();
+
+ @Rule
+ public DistributedDiskDirRule distributedDiskDirRule = new
DistributedDiskDirRule();
+
+ private Properties createLocatorConfiguration(int distributedSystemId, int
localLocatorPort,
+ int remoteLocatorPort) {
+ Properties config = new Properties();
+ config.setProperty(MCAST_PORT, "0");
+ config.setProperty(DISTRIBUTED_SYSTEM_ID,
String.valueOf(distributedSystemId));
+ config.setProperty(LOCATORS, "localhost[" + localLocatorPort + ']');
+ config.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocatorPort +
']');
+ config.setProperty(START_LOCATOR,
+ "localhost[" + localLocatorPort +
"],server=true,peer=true,hostname-for-clients=localhost");
+
+ return config;
+ }
+
+ private Properties createServerConfiguration(int localLocatorPort) {
+ Properties config = new Properties();
+ config.setProperty(MCAST_PORT, "0");
+ config.setProperty(LOCATORS, "localhost[" + localLocatorPort + ']');
+
+ return config;
+ }
+
+ private void createDiskStore() {
+ String basePath = distributedDiskDirRule.getDiskDir().getAbsolutePath();
+ File diskDirectory = new File(basePath + File.separator + DISK_STORE_ID);
+ DiskStoreFactory diskStoreFactory =
cacheRule.getCache().createDiskStoreFactory();
+ diskStoreFactory.setAutoCompact(true);
+ diskStoreFactory.setAllowForceCompaction(true);
+ diskStoreFactory.setDiskDirs(new File[] {diskDirectory});
+ diskStoreFactory.create(DISK_STORE_ID);
+ }
+
+ private void createRegion(RegionShortcut regionShortcut) {
+ cacheRule.getCache()
+ .<String, String>createRegionFactory(regionShortcut)
+ .create(REGION_NAME);
+ }
+
+ private void createGatewayReceiver() {
+ GatewayReceiverFactory gatewayReceiverFactory =
+ cacheRule.getCache().createGatewayReceiverFactory();
+ gatewayReceiverFactory.setManualStart(false);
+ gatewayReceiverFactory.create();
+ }
+
+ private void createGatewaySender(boolean parallel, int
remoteDistributedSystemId) {
+ GatewaySenderFactory gatewaySenderFactory =
cacheRule.getCache().createGatewaySenderFactory();
+ gatewaySenderFactory.setParallel(parallel);
+ gatewaySenderFactory.setDiskSynchronous(true);
+ gatewaySenderFactory.setPersistenceEnabled(true);
+ gatewaySenderFactory.setDiskStoreName(DISK_STORE_ID);
+ gatewaySenderFactory.create(GATEWAY_SENDER_ID, remoteDistributedSystemId);
+ }
+
+ private void createServerWithRegionAndGatewayReceiver(RegionShortcut
regionShortcut) {
+ createGatewayReceiver();
+ createRegion(regionShortcut);
+ }
+
+ private void createServerWithRegionAndPersistentGatewaySender(RegionShortcut
regionShortcut,
+ int remoteDistributedSystemId, boolean parallel) {
+ createDiskStore();
+ createRegion(regionShortcut);
+ createGatewaySender(parallel, remoteDistributedSystemId);
+ Region<String, String> region =
cacheRule.getCache().getRegion(REGION_NAME);
+ region.getAttributesMutator().addGatewaySenderId(GATEWAY_SENDER_ID);
+ }
+
+ private void gracefullyDisconnect() {
+
InternalDistributedSystem.getConnectedInstance().stopReconnectingNoDisconnect();
+ InternalDistributedSystem.getConnectedInstance().disconnect();
+ await()
+ .untilAsserted(() ->
assertThat(InternalDistributedSystem.getConnectedInstance()).isNull());
+ }
+
+ private void awaitForQueueSize(int queueSize) {
+ GatewaySender gatewaySender =
cacheRule.getCache().getGatewaySender(GATEWAY_SENDER_ID);
+ await().untilAsserted(() -> {
+ Set<RegionQueue> queues = ((AbstractGatewaySender)
gatewaySender).getQueues();
+ int totalSize = queues.stream().mapToInt(RegionQueue::size).sum();
+ assertThat(queueSize).isEqualTo(totalSize);
+ });
+ }
+
+ @SuppressWarnings("unused")
+ static Object[] regionAndGatewayTypes() {
+ ArrayList<Object[]> parameters = new ArrayList<>();
+ parameters.add(new Object[] {RegionShortcut.PARTITION, true});
+ parameters.add(new Object[] {RegionShortcut.REPLICATE, false});
+
+ return parameters.toArray();
+ }
+
+ @Before
+ public void setUp() {
+ VM locatorCluster1 = getVM(0);
+ serverCluster1 = getVM(1);
+ VM locatorCluster2 = getVM(2);
+ serverCluster2 = getVM(3);
+
+ int[] ports = getRandomAvailableTCPPortsForDUnitSite(2);
+ site1Port = ports[0];
+ site2Port = ports[1];
+
+ // Start 2 sites, one locator and one server per site.
+ locatorCluster1
+ .invoke(() -> cacheRule.createCache(createLocatorConfiguration(1,
site1Port, site2Port)));
+ locatorCluster2
+ .invoke(() -> cacheRule.createCache(createLocatorConfiguration(2,
site2Port, site1Port)));
+
+ serverCluster1.invoke(() ->
cacheRule.createCache(createServerConfiguration(site1Port)));
+ serverCluster2.invoke(() ->
cacheRule.createCache(createServerConfiguration(site2Port)));
+ }
+
+ /**
+ * The tests executes the following:
+ * - Creates region and gateway-receiver on cluster2.
+ * - Creates the region and gateway-sender on cluster1.
+ * - Populates the region and waits until WAN replication has finished.
+ * - Restarts server on cluster1, and stops it afterwards (the initial
compaction occurs during
+ * startup and the disk validation is done offline).
+ * - Asserts that there are no orphaned drf files, neither compact-able
records on the disks-tore.
+ */
+ @Test
+ @TestCaseName(TEST_CASE_NAME)
+ @Parameters(method = "regionAndGatewayTypes")
+ public void
diskStoreShouldBeCompactedOnMemberRestartWhenAllEventsHaveBeenDispatched(
+ RegionShortcut regionShortcut, boolean parallel) throws Exception {
+ final int entries = 100;
Review comment:
I made them different on purpose, the amount of entries is irrelevant
and I didn't want to "hardcode" it as a static field within the test class as
we might want to add extra tests (with specific requirements about the amount
of entries to insert) to this class in the future.
----------------------------------------------------------------
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:
us...@infra.apache.org
> java.lang.IllegalArgumentException: Too large (805306401 expected elements
> with load factor 0.75)
> -------------------------------------------------------------------------------------------------
>
> Key: GEODE-8029
> URL: https://issues.apache.org/jira/browse/GEODE-8029
> Project: Geode
> Issue Type: Bug
> Components: configuration, core, gfsh
> Affects Versions: 1.9.0
> Reporter: Jagadeesh sivasankaran
> Assignee: Juan Ramos
> Priority: Major
> Labels: GeodeCommons, caching-applications
> Attachments: Screen Shot 2020-04-27 at 12.21.19 PM.png, Screen Shot
> 2020-04-27 at 12.21.19 PM.png, server02.log
>
>
> we have a cluster of three Locator Geode and three Cache Server running in
> CentOS servers. Today (April 27) after patching our CENTOS servers , all
> locator and 2 servers came up , But one Cache server was not starting . here
> is the Exception details. Please let me know how to resolve the beloe issue
> and need any configuration changes to diskstore ?
>
>
> Starting a Geode Server in /app/provServerHO2...
> ....................................................................................................................................................................................................................The
> Cache Server process terminated unexpectedly with exit status 1. Please
> refer to the log file in /app/provServerHO2 for full details.
> Exception in thread "main" java.lang.IllegalArgumentException: Too large
> (805306401 expected elements with load factor 0.75)
> at it.unimi.dsi.fastutil.HashCommon.arraySize(HashCommon.java:222)
> at it.unimi.dsi.fastutil.ints.IntOpenHashSet.add(IntOpenHashSet.java:308)
> at
> org.apache.geode.internal.cache.DiskStoreImpl$OplogEntryIdSet.add(DiskStoreImpl.java:3474)
> at org.apache.geode.internal.cache.Oplog.readDelEntry(Oplog.java:3007)
> at org.apache.geode.internal.cache.Oplog.recoverDrf(Oplog.java:1500)
> at
> org.apache.geode.internal.cache.PersistentOplogSet.recoverOplogs(PersistentOplogSet.java:445)
> at
> org.apache.geode.internal.cache.PersistentOplogSet.recoverRegionsThatAreReady(PersistentOplogSet.java:369)
> at
> org.apache.geode.internal.cache.DiskStoreImpl.recoverRegionsThatAreReady(DiskStoreImpl.java:2053)
> at
> org.apache.geode.internal.cache.DiskStoreImpl.initializeIfNeeded(DiskStoreImpl.java:2041)
> security-peer-auth-init=
> at
> org.apache.geode.internal.cache.DiskStoreImpl.doInitialRecovery(DiskStoreImpl.java:2046)
> at
> org.apache.geode.internal.cache.DiskStoreFactoryImpl.initializeDiskStore(DiskStoreFactoryImpl.java:184)
> at
> org.apache.geode.internal.cache.DiskStoreFactoryImpl.create(DiskStoreFactoryImpl.java:150)
> at
> org.apache.geode.internal.cache.xmlcache.CacheCreation.createDiskStore(CacheCreation.java:794)
> at
> org.apache.geode.internal.cache.xmlcache.CacheCreation.initializePdxDiskStore(CacheCreation.java:785)
> at
> org.apache.geode.internal.cache.xmlcache.CacheCreation.create(CacheCreation.java:509)
> at
> org.apache.geode.internal.cache.xmlcache.CacheXmlParser.create(CacheXmlParser.java:337)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.loadCacheXml(GemFireCacheImpl.java:4272)
> at
> org.apache.geode.internal.cache.ClusterConfigurationLoader.applyClusterXmlConfiguration(ClusterConfigurationLoader.java:197)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.applyJarAndXmlFromClusterConfig(GemFireCacheImpl.java:1240)
> at
> org.apache.geode.internal.cache.GemFireCacheImpl.initialize(GemFireCacheImpl.java:1206)
> at
> org.apache.geode.internal.cache.InternalCacheBuilder.create(InternalCacheBuilder.java:207)
> at
> org.apache.geode.internal.cache.InternalCacheBuilder.create(InternalCacheBuilder.java:164)
> at org.apache.geode.cache.CacheFactory.create(CacheFactory.java:139)
> at
> org.apache.geode.distributed.internal.DefaultServerLauncherCacheProvider.createCache(DefaultServerLauncherCacheProvider.java:52)
> at
> org.apache.geode.distributed.ServerLauncher.createCache(ServerLauncher.java:869)
> at org.apache.geode.distributed.ServerLauncher.start(ServerLauncher.java:786)
> at org.apache.geode.distributed.ServerLauncher.run(ServerLauncher.java:716)
> at org.apache.geode.distributed.ServerLauncher.main(ServerLauncher.java:236)
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
