| /* |
| * 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.cache30; |
| |
| import static java.lang.System.out; |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static java.util.concurrent.TimeUnit.MINUTES; |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| import static org.apache.geode.cache.DataPolicy.REPLICATE; |
| import static org.apache.geode.cache.LossAction.RECONNECT; |
| import static org.apache.geode.cache.ResumptionAction.NONE; |
| import static org.apache.geode.cache.Scope.DISTRIBUTED_ACK; |
| import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE; |
| import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_AUTO_RECONNECT; |
| import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION; |
| import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL; |
| import static org.apache.geode.distributed.ConfigurationProperties.MAX_NUM_RECONNECT_TRIES; |
| import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.distributed.ConfigurationProperties.MEMBER_TIMEOUT; |
| import static org.apache.geode.distributed.ConfigurationProperties.ROLES; |
| import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER; |
| import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR; |
| import static org.apache.geode.distributed.Locator.getLocator; |
| import static org.apache.geode.distributed.internal.membership.gms.MembershipManagerHelper.getMembershipManager; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.await; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout; |
| import static org.apache.geode.test.dunit.Host.getHost; |
| import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException; |
| import static org.apache.geode.test.dunit.ThreadUtils.join; |
| import static org.assertj.core.api.Assertions.assertThatThrownBy; |
| import static org.hamcrest.Matchers.notNullValue; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNotSame; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.io.File; |
| import java.io.FileWriter; |
| import java.io.IOException; |
| import java.io.PrintWriter; |
| import java.util.Iterator; |
| import java.util.Properties; |
| import java.util.Set; |
| |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.CacheFactory; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.Declarable; |
| import org.apache.geode.cache.LossAction; |
| import org.apache.geode.cache.MembershipAttributes; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionAttributes; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.RegionEvent; |
| import org.apache.geode.cache.RegionExistsException; |
| import org.apache.geode.cache.ResumptionAction; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.TimeoutException; |
| import org.apache.geode.cache.util.CacheListenerAdapter; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.distributed.DistributedSystemDisconnectedException; |
| import org.apache.geode.distributed.Locator; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem.ReconnectListener; |
| import org.apache.geode.distributed.internal.InternalLocator; |
| import org.apache.geode.distributed.internal.ServerLocator; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.distributed.internal.membership.gms.MembershipManagerHelper; |
| import org.apache.geode.distributed.internal.membership.gms.mgr.GMSMembershipManager; |
| import org.apache.geode.examples.SimpleSecurityManager; |
| import org.apache.geode.internal.AvailablePort; |
| import org.apache.geode.internal.AvailablePortHelper; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator; |
| import org.apache.geode.test.awaitility.GeodeAwaitility; |
| import org.apache.geode.test.dunit.Assert; |
| import org.apache.geode.test.dunit.AsyncInvocation; |
| import org.apache.geode.test.dunit.DistributedTestUtils; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.IgnoredException; |
| import org.apache.geode.test.dunit.Invoke; |
| import org.apache.geode.test.dunit.SerializableCallable; |
| import org.apache.geode.test.dunit.SerializableRunnable; |
| import org.apache.geode.test.dunit.ThreadUtils; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.Wait; |
| import org.apache.geode.test.dunit.WaitCriterion; |
| import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase; |
| import org.apache.geode.test.junit.categories.MembershipTest; |
| |
| @SuppressWarnings("serial") |
| @Category({MembershipTest.class}) |
| public class ReconnectDUnitTest extends JUnit4CacheTestCase { |
| static int locatorPort; |
| static Locator locator; |
| static DistributedSystem savedSystem; |
| static GemFireCacheImpl savedCache; |
| static int locatorVMNumber = 3; |
| static Thread gfshThread; |
| |
| static Properties dsProperties; |
| static String fileSeparator = File.separator; |
| |
| public ReconnectDUnitTest() { |
| super(); |
| } |
| |
| @Override |
| public final void postSetUp() throws Exception { |
| locatorPort = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET); |
| final int locPort = locatorPort; |
| Host.getHost(0).getVM(locatorVMNumber).invoke(new SerializableRunnable("start locator") { |
| @Override |
| public void run() { |
| try { |
| disconnectFromDS(); |
| dsProperties = null; |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| locator = Locator.startLocatorAndDS(locatorPort, new File(""), props); |
| system = (InternalDistributedSystem) locator.getDistributedSystem(); |
| cache = ((InternalLocator) locator).getCache(); |
| ReconnectDUnitTest.savedSystem = locator.getDistributedSystem(); |
| IgnoredException.addIgnoredException( |
| "org.apache.geode.ForcedDisconnectException||Possible loss of quorum"); |
| // MembershipManagerHelper.getMembershipManager(InternalDistributedSystem.getConnectedInstance()).setDebugJGroups(true); |
| } catch (IOException e) { |
| Assert.fail("unable to start locator", e); |
| } |
| } |
| }); |
| |
| SerializableRunnable setDistributedSystemProperties = |
| new SerializableRunnable("set distributed system properties") { |
| @Override |
| public void run() { |
| dsProperties = null; |
| locatorPort = locPort; |
| getDistributedSystemProperties(); |
| } |
| }; |
| setDistributedSystemProperties.run(); |
| Invoke.invokeInEveryVM(setDistributedSystemProperties); |
| |
| beginCacheXml(); |
| createRegion("myRegion", createAtts()); |
| finishCacheXml("MyDisconnect"); |
| // Cache cache = getCache(); |
| closeCache(); |
| basicGetSystem().disconnect(); |
| } |
| |
| @Override |
| public Properties getDistributedSystemProperties() { |
| if (dsProperties == null) { |
| dsProperties = new Properties(); |
| dsProperties.put(MAX_WAIT_TIME_RECONNECT, "20000"); |
| dsProperties.put(ENABLE_NETWORK_PARTITION_DETECTION, "true"); |
| dsProperties.put(DISABLE_AUTO_RECONNECT, "false"); |
| dsProperties.put(ENABLE_CLUSTER_CONFIGURATION, "false"); |
| dsProperties.put(LOCATORS, "localHost[" + locatorPort + "]"); |
| dsProperties.put(MCAST_PORT, "0"); |
| dsProperties.put(MEMBER_TIMEOUT, "1000"); |
| dsProperties.put(LOG_LEVEL, "info"); |
| dsProperties.put(SECURITY_MANAGER, SimpleSecurityManager.class.getName()); |
| dsProperties.put("security-username", "clusterManage"); |
| dsProperties.put("security-password", "clusterManage"); |
| addDSProps(dsProperties); |
| } |
| return dsProperties; |
| } |
| |
| @Override |
| public final void postTearDownCacheTestCase() throws Exception { |
| System.out.println("entering postTearDownCacheTestCase"); |
| SerializableRunnable disconnect = new SerializableRunnable("disconnect and clean up") { |
| @Override |
| public void run() { |
| if (savedSystem != null && savedSystem.isReconnecting()) { |
| savedSystem.stopReconnecting(); |
| } |
| savedSystem = null; |
| savedCache = null; |
| dsProperties = null; |
| locator = null; |
| locatorPort = 0; |
| } |
| }; |
| Invoke.invokeInEveryVM(disconnect); |
| disconnect.run(); |
| disconnectAllFromDS(); |
| } |
| |
| /** |
| * Creates some region attributes for the regions being created. |
| */ |
| private RegionAttributes createAtts() { |
| AttributesFactory factory = new AttributesFactory(); |
| |
| { |
| factory.setDataPolicy(DataPolicy.REPLICATE); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| } |
| |
| return factory.create(); |
| } |
| |
| |
| @Test |
| public void testReconnectWithQuorum() throws Exception { |
| // quorum check fails, then succeeds |
| IgnoredException.addIgnoredException("killing member's ds"); |
| Host host = Host.getHost(0); |
| VM vm0 = host.getVM(0); |
| VM vm1 = host.getVM(1); |
| VM locatorVm = host.getVM(locatorVMNumber); |
| |
| final int locPort = locatorPort; |
| |
| final String xmlFileLoc = (new File(".")).getAbsolutePath(); |
| |
| // disable disconnects in the locator so we have some stability |
| locatorVm.invoke(new SerializableRunnable("disable force-disconnect") { |
| @Override |
| public void run() { |
| GMSMembershipManager mgr = (GMSMembershipManager) MembershipManagerHelper |
| .getMembershipManager(Locator.getLocator().getDistributedSystem()); |
| mgr.disableDisconnectOnQuorumLossForTesting(); |
| } |
| }); |
| |
| SerializableCallable create = |
| new SerializableCallable("Create Cache and Regions from cache.xml") { |
| @Override |
| public Object call() throws CacheException { |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(CACHE_XML_FILE, xmlFileLoc + fileSeparator + "MyDisconnect-cache.xml"); |
| props.put(MAX_NUM_RECONNECT_TRIES, "2"); |
| // props.put("log-file", "autoReconnectVM"+VM.getCurrentVMNum()+"_"+getPID()+".log"); |
| cache = (InternalCache) new CacheFactory(props).create(); |
| IgnoredException.addIgnoredException( |
| "org.apache.geode.ForcedDisconnectException||Possible loss of quorum"); |
| Region myRegion = cache.getRegion("root/myRegion"); |
| ReconnectDUnitTest.savedSystem = cache.getDistributedSystem(); |
| myRegion.put("MyKey1", "MyValue1"); |
| return savedSystem.getDistributedMember(); |
| } |
| }; |
| |
| System.out.println("creating caches in vm0 and vm1"); |
| vm0.invoke(create); |
| vm1.invoke(create); |
| |
| // view is [locator(3), vm0(15), vm1(10), vm2(10)] |
| |
| /* |
| * now we want to kick out the locator and observe that it reconnects using its rebooted |
| * location service |
| */ |
| System.out.println("disconnecting locator"); |
| forceDisconnect(locatorVm); |
| waitForReconnect(locatorVm); |
| |
| // if the locator reconnected it did so with its own location |
| // service since it doesn't know about any other locators |
| ensureLocationServiceRunning(locatorVm); |
| |
| } |
| |
| @Test |
| public void testReconnectOnForcedDisconnect() throws Exception { |
| doTestReconnectOnForcedDisconnect(false); |
| } |
| |
| /** bug #51335 - customer is also trying to recreate the cache */ |
| // this test is disabled due to a high failure rate during CI test runs. |
| // see bug #52160 |
| @Test |
| public void testReconnectCollidesWithApplication() throws Exception { |
| doTestReconnectOnForcedDisconnect(true); |
| } |
| |
| public void doTestReconnectOnForcedDisconnect(final boolean createInAppToo) throws Exception { |
| |
| IgnoredException.addIgnoredException("killing member's ds"); |
| |
| Host host = Host.getHost(0); |
| VM vm0 = host.getVM(0); |
| VM vm1 = host.getVM(1); |
| |
| final int locPort = locatorPort; |
| final int secondLocPort = AvailablePortHelper.getRandomAvailableTCPPort(); |
| |
| DistributedTestUtils.deleteLocatorStateFile(locPort, secondLocPort); |
| |
| |
| final String xmlFileLoc = (new File(".")).getAbsolutePath(); |
| |
| SerializableCallable create1 = |
| new SerializableCallable("Create Cache and Regions from cache.xml") { |
| @Override |
| public Object call() throws CacheException { |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(CACHE_XML_FILE, xmlFileLoc + fileSeparator + "MyDisconnect-cache.xml"); |
| props.put(MAX_WAIT_TIME_RECONNECT, "1000"); |
| cache = (InternalCache) new CacheFactory(props).create(); |
| Region myRegion = cache.getRegion("root/myRegion"); |
| ReconnectDUnitTest.savedSystem = cache.getDistributedSystem(); |
| myRegion.put("MyKey1", "MyValue1"); |
| return savedSystem.getDistributedMember(); |
| } |
| }; |
| |
| SerializableCallable create2 = |
| new SerializableCallable("Create Cache and Regions from cache.xml") { |
| @Override |
| public Object call() throws CacheException { |
| locatorPort = locPort; |
| final Properties props = getDistributedSystemProperties(); |
| props.put(CACHE_XML_FILE, xmlFileLoc + fileSeparator + "MyDisconnect-cache.xml"); |
| props.put(MAX_WAIT_TIME_RECONNECT, "5000"); |
| props.put(START_LOCATOR, "localhost[" + secondLocPort + "]"); |
| props.put(LOCATORS, props.get(LOCATORS) + ",localhost[" + secondLocPort + "]"); |
| getSystem(props); |
| cache = getCache(); |
| ReconnectDUnitTest.savedSystem = cache.getDistributedSystem(); |
| Region myRegion = cache.getRegion("root/myRegion"); |
| myRegion.put("Mykey2", "MyValue2"); |
| assertNotNull(myRegion.get("MyKey1")); |
| if (createInAppToo) { |
| Thread recreateCacheThread = new Thread("ReconnectDUnitTest.createInAppThread") { |
| @Override |
| public void run() { |
| while (!cache.isClosed()) { |
| Wait.pause(100); |
| } |
| try { |
| cache = (InternalCache) new CacheFactory(props).create(); |
| System.err.println( |
| "testReconnectCollidesWithApplication failed - application thread was able to create a cache"); |
| } catch (IllegalStateException cacheExists) { |
| // expected |
| } |
| } |
| }; |
| recreateCacheThread.setDaemon(true); |
| recreateCacheThread.start(); |
| } |
| return cache.getDistributedSystem().getDistributedMember(); |
| } |
| }; |
| |
| vm0.invoke(create1); |
| final DistributedMember dm = (DistributedMember) vm1.invoke(create2); |
| |
| IgnoredException.addIgnoredException("ForcedDisconnectException"); |
| forceDisconnect(vm1); |
| |
| DistributedMember newdm = |
| (DistributedMember) vm1.invoke(new SerializableCallable("wait for reconnect(1)") { |
| @Override |
| public Object call() { |
| final DistributedSystem ds = savedSystem; |
| savedSystem = null; |
| GeodeAwaitility.await().untilAsserted(new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return ds.isReconnecting(); |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for ds to begin reconnecting"; |
| } |
| }); |
| System.out.println("entering reconnect wait for " + ds); |
| System.out.println("ds.isReconnecting() = " + ds.isReconnecting()); |
| boolean failure = true; |
| try { |
| ds.waitUntilReconnected(getTimeout().getValueInMS(), MILLISECONDS); |
| savedSystem = ds.getReconnectedSystem(); |
| locator = (InternalLocator) getLocator(); |
| assertTrue("Expected system to be restarted", ds.getReconnectedSystem() != null); |
| assertTrue("Expected system to be running", ds.getReconnectedSystem().isConnected()); |
| assertTrue("Expected there to be a locator", locator != null); |
| assertTrue("Expected locator to be restarted", |
| !((InternalLocator) locator).isStopped()); |
| failure = false; |
| cache = ((InternalLocator) locator).getCache(); |
| system = cache.getInternalDistributedSystem(); |
| assertTrue( |
| ((GMSMembershipManager) getMembershipManager(system)) |
| .getServices().getMessenger().isOldMembershipIdentifier(dm)); |
| return ds.getReconnectedSystem().getDistributedMember(); |
| } catch (InterruptedException e) { |
| System.err.println("interrupted while waiting for reconnect"); |
| return null; |
| } finally { |
| if (failure) { |
| ds.disconnect(); |
| } |
| } |
| } |
| }); |
| assertNotSame(dm, newdm); |
| // force another reconnect and show that stopReconnecting works |
| forceDisconnect(vm1); |
| boolean stopped = (Boolean) vm1.invoke(new SerializableCallable("wait for reconnect and stop") { |
| @Override |
| public Object call() { |
| final DistributedSystem ds = savedSystem; |
| savedSystem = null; |
| GeodeAwaitility.await().untilAsserted(new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return ds.isReconnecting() || ds.getReconnectedSystem() != null; |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for reconnect to commence in " + ds; |
| } |
| |
| }); |
| ds.stopReconnecting(); |
| assertFalse(ds.isReconnecting()); |
| DistributedSystem newDs = ds.getReconnectedSystem(); |
| if (newDs != null) { |
| System.err.println("expected distributed system to be disconnected: " + newDs); |
| newDs.disconnect(); |
| return false; |
| } |
| return true; |
| } |
| }); |
| assertTrue("expected DistributedSystem to disconnect", stopped); |
| |
| // recreate the system in vm1 without a locator and crash it |
| DistributedMember evenNewerdm = (DistributedMember) vm1.invoke(create1); |
| forceDisconnect(vm1); |
| newdm = waitForReconnect(vm1); |
| assertNotSame("expected a reconnect to occur in member", evenNewerdm, newdm); |
| DistributedTestUtils.deleteLocatorStateFile(locPort); |
| DistributedTestUtils.deleteLocatorStateFile(secondLocPort); |
| } |
| |
| private DistributedMember getDMID(VM vm) { |
| return (DistributedMember) vm.invoke(new SerializableCallable("get ID") { |
| @Override |
| public Object call() { |
| ReconnectDUnitTest.savedSystem = cache.getDistributedSystem(); |
| return ReconnectDUnitTest.savedSystem.getDistributedMember(); |
| } |
| }); |
| } |
| |
| /** this will throw an exception if location services aren't running */ |
| private void ensureLocationServiceRunning(VM vm) { |
| vm.invoke("ensureLocationServiceRunning", () -> { |
| await().untilAsserted(() -> { |
| InternalLocator intloc = (InternalLocator) locator; |
| ServerLocator serverLocator = intloc.getServerLocatorAdvisee(); |
| // the initialization flag in the locator's ControllerAdvisor will |
| // be set if a handshake has been performed |
| assertTrue(serverLocator.getDistributionAdvisor().isInitialized()); |
| }); |
| }); |
| } |
| |
| private DistributedMember waitForReconnect(VM vm) { |
| return (DistributedMember) vm |
| .invoke(new SerializableCallable("wait for Reconnect and return ID") { |
| @Override |
| public Object call() { |
| out.println("waitForReconnect invoked"); |
| final DistributedSystem ds = savedSystem; |
| savedSystem = null; |
| GeodeAwaitility.await().untilAsserted(new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return ds.isReconnecting(); |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for ds to begin reconnecting"; |
| } |
| }); |
| long waitTime = 600; |
| System.out.println("VM" + VM.getCurrentVMNum() + " waiting up to " |
| + waitTime + " seconds for reconnect to complete"); |
| try { |
| ds.waitUntilReconnected(waitTime, SECONDS); |
| } catch (InterruptedException e) { |
| fail("interrupted while waiting for reconnect"); |
| } |
| assertTrue("expected system to be reconnected", ds.getReconnectedSystem() != null); |
| int oldViewId = |
| getMembershipManager(ds).getLocalMember().getVmViewId(); |
| int newViewId = |
| ((InternalDistributedMember) ds.getReconnectedSystem().getDistributedMember()) |
| .getVmViewId(); |
| if (!(newViewId > oldViewId)) { |
| fail("expected a new ID to be assigned. oldViewId=" + oldViewId + "; newViewId=" |
| + newViewId); |
| } |
| return ds.getReconnectedSystem().getDistributedMember(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testReconnectALocator() throws Exception { |
| Host host = Host.getHost(0); |
| VM vm0 = host.getVM(0); |
| VM vm1 = host.getVM(1); |
| VM locatorVm = host.getVM(3); |
| DistributedMember dm, newdm; |
| |
| final int locPort = locatorPort; |
| final int secondLocPort = AvailablePortHelper.getRandomAvailableTCPPort(); |
| |
| DistributedTestUtils.deleteLocatorStateFile(locPort, secondLocPort); |
| |
| final String xmlFileLoc = (new File(".")).getAbsolutePath(); |
| |
| // This locator was started in setUp. |
| File locatorViewLog = |
| new File(locatorVm.getWorkingDirectory(), "locator" + locatorPort + "views.log"); |
| assertTrue("Expected to find " + locatorViewLog.getPath() + " file", locatorViewLog.exists()); |
| long logSize = locatorViewLog.length(); |
| |
| vm0.invoke("Create a second locator", () -> { |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(MAX_WAIT_TIME_RECONNECT, "1000"); |
| props.put(LOCATORS, props.get(LOCATORS) + ",localhost[" + locPort + "]"); |
| props.put(ENABLE_CLUSTER_CONFIGURATION, "false"); |
| try { |
| InternalLocator locator = |
| (InternalLocator) Locator.startLocatorAndDS(secondLocPort, null, props); |
| system = (InternalDistributedSystem) locator.getDistributedSystem(); |
| cache = locator.getCache(); |
| } catch (IOException e) { |
| Assert.fail("exception starting locator", e); |
| } |
| }); |
| |
| File locator2ViewLog = |
| new File(vm0.getWorkingDirectory(), "locator" + secondLocPort + "views.log"); |
| assertTrue("Expected to find " + locator2ViewLog.getPath() + " file", locator2ViewLog.exists()); |
| long log2Size = locator2ViewLog.length(); |
| |
| // create a cache in vm1 so there is more weight in the system |
| vm1.invoke("Create Cache and Regions from cache.xml", () -> { |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(CACHE_XML_FILE, xmlFileLoc + fileSeparator + "MyDisconnect-cache.xml"); |
| props.put(MAX_WAIT_TIME_RECONNECT, "1000"); |
| ReconnectDUnitTest.savedSystem = getSystem(props); |
| cache = getCache(); |
| Region myRegion = cache.getRegion("root/myRegion"); |
| myRegion.put("MyKey1", "MyValue1"); |
| return savedSystem.getDistributedMember(); |
| }); |
| |
| try { |
| dm = getDMID(vm0); |
| createGfshWaitingThread(vm0); |
| forceDisconnect(vm0); |
| newdm = waitForReconnect(vm0); |
| assertGfshWaitingThreadAlive(vm0); |
| |
| assertTrue("Expected the restarted member to be hosting a running locator", |
| vm0.invoke("check for running locator", () -> { |
| await("waiting for locator to restart") |
| .until(Locator::getLocator, notNullValue()); |
| if (((InternalLocator) Locator.getLocator()).isStopped()) { |
| System.err.println("found a stopped locator"); |
| return false; |
| } |
| return true; |
| })); |
| |
| assertNotSame("expected a reconnect to occur in the locator", dm, newdm); |
| |
| // the log should have been opened and appended with a new view |
| assertTrue("expected " + locator2ViewLog.getPath() + " to grow in size", |
| locator2ViewLog.length() > log2Size); |
| // the other locator should have logged a new view |
| assertTrue("expected " + locatorViewLog.getPath() + " to grow in size", |
| locatorViewLog.length() > logSize); |
| |
| } finally { |
| vm0.invoke(new SerializableRunnable("stop locator") { |
| @Override |
| public void run() { |
| Locator loc = Locator.getLocator(); |
| if (loc != null) { |
| loc.stop(); |
| } |
| if (gfshThread != null && gfshThread.isAlive()) { |
| gfshThread.interrupt(); |
| } |
| gfshThread = null; |
| } |
| }); |
| DistributedTestUtils.deleteLocatorStateFile(locPort); |
| DistributedTestUtils.deleteLocatorStateFile(secondLocPort); |
| } |
| } |
| |
| @SuppressWarnings("serial") |
| private void createGfshWaitingThread(VM vm) { |
| vm.invoke(new SerializableRunnable("create Gfsh-like waiting thread") { |
| @Override |
| public void run() { |
| final Locator loc = Locator.getLocator(); |
| assertNotNull(loc); |
| gfshThread = new Thread("ReconnectDUnitTest_Gfsh_thread") { |
| @Override |
| public void run() { |
| try { |
| ((InternalLocator) loc).waitToStop(); |
| } catch (InterruptedException e) { |
| System.out.println(Thread.currentThread().getName() + " interrupted - exiting"); |
| } |
| } |
| }; |
| gfshThread.setDaemon(true); |
| gfshThread.start(); |
| System.out.println("created gfsh thread: " + gfshThread); |
| } |
| }); |
| } |
| |
| @SuppressWarnings("serial") |
| private void assertGfshWaitingThreadAlive(VM vm) { |
| vm.invoke(new SerializableRunnable("assert gfshThread is still waiting") { |
| @Override |
| public void run() { |
| assertTrue(gfshThread.isAlive()); |
| } |
| }); |
| } |
| |
| /** |
| * Test the reconnect behavior when the required roles are missing. Reconnect is triggered as a |
| * Reliability policy. The test is to see if the reconnect is triggered for the configured number |
| * of times |
| */ |
| |
| @Test |
| public void testReconnectWithRoleLoss() throws TimeoutException, RegionExistsException { |
| |
| final String rr1 = "RoleA"; |
| final String rr2 = "RoleB"; |
| final String[] requiredRoles = {rr1, rr2}; |
| final int locPort = locatorPort; |
| final String xmlFileLoc = (new File(".")).getAbsolutePath(); |
| |
| |
| beginCacheXml(); |
| |
| locatorPort = locPort; |
| Properties config = getDistributedSystemProperties(); |
| config.put(ROLES, ""); |
| config.put(LOG_LEVEL, "info"); |
| // config.put("log-file", "roleLossController.log"); |
| // creating the DS |
| getSystem(config); |
| |
| MembershipAttributes ra = |
| new MembershipAttributes(requiredRoles, LossAction.RECONNECT, ResumptionAction.NONE); |
| |
| AttributesFactory fac = new AttributesFactory(); |
| fac.setMembershipAttributes(ra); |
| fac.setScope(Scope.DISTRIBUTED_ACK); |
| |
| RegionAttributes attr = fac.create(); |
| createRootRegion("MyRegion", attr); |
| |
| // writing the cachexml file. |
| |
| File file = new File("RoleReconnect-cache.xml"); |
| try { |
| PrintWriter pw = new PrintWriter(new FileWriter(file), true); |
| CacheXmlGenerator.generate(getCache(), pw); |
| pw.close(); |
| } catch (IOException ex) { |
| Assert.fail("IOException during cache.xml generation to " + file, ex); |
| } |
| closeCache(); |
| basicGetSystem().disconnect(); |
| |
| System.out.println("disconnected from the system..."); |
| Host host = Host.getHost(0); |
| |
| VM vm0 = host.getVM(0); |
| |
| // Recreating from the cachexml. |
| |
| SerializableRunnable roleLoss = new CacheSerializableRunnable("ROLERECONNECTTESTS") { |
| @Override |
| public void run2() throws RuntimeException { |
| System.out.println("####### STARTING THE REAL TEST ##########"); |
| |
| locatorPort = locPort; |
| dsProperties = null; |
| Properties props = getDistributedSystemProperties(); |
| props.put(CACHE_XML_FILE, xmlFileLoc + fileSeparator + "RoleReconnect-cache.xml"); |
| props.put(MAX_WAIT_TIME_RECONNECT, "200"); |
| final int timeReconnect = 3; |
| props.put(MAX_NUM_RECONNECT_TRIES, "3"); |
| props.put(LOG_LEVEL, "info"); |
| // props.put("log-file", "roleLossVM0.log"); |
| |
| getSystem(props); |
| |
| addReconnectListener(); |
| |
| IgnoredException.addIgnoredException("CacheClosedException"); |
| try { |
| getCache(props); |
| throw new RuntimeException("The test should throw a CancelException "); |
| } catch (CancelException ignor) { // can be caused by role loss during intialization. |
| System.out.println("Got Expected CancelException "); |
| } |
| |
| WaitCriterion ev = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return reconnectTries >= timeReconnect; |
| } |
| |
| @Override |
| public String description() { |
| return "Waiting for reconnect count " + timeReconnect + " currently " + reconnectTries; |
| } |
| }; |
| Wait.waitForCriterion(ev, 60 * 1000, 200, true); |
| assertEquals(timeReconnect, reconnectTries); |
| } |
| |
| }; |
| |
| vm0.invoke(roleLoss); |
| |
| |
| } |
| |
| |
| |
| public static volatile int reconnectTries; |
| |
| public static volatile boolean initialized = false; |
| |
| public static volatile boolean initialRolePlayerStarted = false; |
| |
| // public static boolean rPut; |
| public static int reconnectTries() { |
| return reconnectTries; |
| } |
| |
| public static boolean isInitialized() { |
| return initialized; |
| } |
| |
| public static boolean isInitialRolePlayerStarted() { |
| return initialRolePlayerStarted; |
| } |
| |
| @Before |
| public void initStatics() { |
| Invoke.invokeInEveryVM(() -> { |
| reconnectTries = 0; |
| initialized = false; |
| initialRolePlayerStarted = false; |
| }); |
| } |
| |
| // See #50944 before enabling the test. This ticket has been closed with wontFix |
| // for the 2014 8.0 release. |
| @Test |
| public void testReconnectWithRequiredRoleRegained() throws Throwable { |
| |
| final String rr1 = "RoleA"; |
| // final String rr2 = "RoleB"; |
| final String[] requiredRoles = {rr1}; |
| // final boolean receivedPut[] = new boolean[1]; |
| |
| final Integer[] numReconnect = new Integer[1]; |
| numReconnect[0] = new Integer(-1); |
| final String myKey = "MyKey"; |
| final String myValue = "MyValue"; |
| final String regionName = "MyRegion"; |
| final int locPort = locatorPort; |
| |
| // CREATE XML FOR MEMBER THAT WILL SEE ROLE LOSS (in this VM) |
| beginCacheXml(); |
| |
| locatorPort = locPort; |
| Properties config = getDistributedSystemProperties(); |
| config.put(ROLES, ""); |
| config.put(LOG_LEVEL, "info"); |
| // creating the DS |
| getSystem(config); |
| |
| MembershipAttributes ra = |
| new MembershipAttributes(requiredRoles, RECONNECT, NONE); |
| |
| AttributesFactory fac = new AttributesFactory(); |
| fac.setMembershipAttributes(ra); |
| fac.setScope(DISTRIBUTED_ACK); |
| fac.setDataPolicy(REPLICATE); |
| |
| RegionAttributes attr = fac.create(); |
| createRootRegion(regionName, attr); |
| |
| // writing the cachexml file. |
| |
| File file = new File("RoleRegained.xml"); |
| try { |
| PrintWriter pw = new PrintWriter(new FileWriter(file), true); |
| CacheXmlGenerator.generate(getCache(), pw); |
| pw.close(); |
| } catch (IOException ex) { |
| Assert.fail("IOException during cache.xml generation to " + file, ex); |
| } |
| closeCache(); |
| // disconnectFromDS(); |
| getSystem().disconnect(); |
| |
| // ################################################################### // |
| // |
| Host host = getHost(0); |
| final VM vm0 = host.getVM(0); |
| final VM vm1 = host.getVM(1); |
| |
| vm0.invoke(new CacheSerializableRunnable("reset reconnect count") { |
| @Override |
| public void run2() throws CacheException { |
| reconnectTries = 0; |
| } |
| }); |
| |
| SerializableRunnable roleAPlayerForCacheInitialization = |
| getRoleAPlayerForCacheInitializationRunnable(vm0, locPort, regionName, |
| "starting roleAplayer, which will initialize, wait for " |
| + "vm0 to initialize, and then close its cache to cause role loss"); |
| AsyncInvocation avkVm1 = vm1.invokeAsync(roleAPlayerForCacheInitialization); |
| |
| CacheSerializableRunnable roleLoss = |
| getRoleLossRunnable(vm1, locPort, regionName, myKey, myValue, |
| "starting role loss vm. When the role is lost it will start" + " trying to reconnect", |
| file.getAbsolutePath()); |
| final AsyncInvocation roleLossAsync = vm0.invokeAsync(roleLoss); |
| |
| System.out.println("waiting for role loss vm to start reconnect attempts"); |
| |
| WaitCriterion ev = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| if (!roleLossAsync.isAlive()) { |
| return true; |
| } |
| int tries = vm0.invoke(() -> ReconnectDUnitTest.reconnectTries()); |
| return tries != 0; |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for event"; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(ev); |
| |
| VM vm2 = host.getVM(2); |
| if (roleLossAsync.isAlive()) { |
| |
| SerializableRunnable roleAPlayer = getRoleAPlayerRunnable(locPort, regionName, myKey, myValue, |
| "starting roleAPlayer in a different vm." |
| + " After this reconnect should succeed in vm0"); |
| |
| vm2.invoke(roleAPlayer); |
| |
| // long startTime = System.currentTimeMillis(); |
| /* |
| * while (numReconnect[0].intValue() > 0){ if((System.currentTimeMillis()-startTime )> 120000) |
| * fail("The test failed because the required role not satisfied" + |
| * "and the number of reconnected tried is not set to zero for " + "more than 2 mins"); try{ |
| * Thread.sleep(15); }catch(Exception ee){ getLogWriter().severe("Exception : "+ee); } } |
| */ |
| System.out.println("waiting for vm0 to finish reconnecting"); |
| ThreadUtils.join(roleLossAsync, 120 * 1000); |
| } |
| |
| if (roleLossAsync.getException() != null) { |
| throw roleLossAsync.getException(); |
| } |
| |
| join(avkVm1, 30 * 1000); |
| if (avkVm1.getException() != null) { |
| Assert.fail("Exception in Vm1", avkVm1.getException()); |
| } |
| |
| } |
| |
| private CacheSerializableRunnable getRoleLossRunnable(final VM otherVM, final int locPort, |
| final String regionName, final String myKey, final Object myValue, |
| final String startupMessage, final String xmlFilePath) { |
| |
| return new CacheSerializableRunnable("roleloss runnable") { |
| @Override |
| public void run2() { |
| System.out.println(startupMessage); |
| WaitCriterion ev = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return otherVM.invoke(() -> ReconnectDUnitTest.isInitialRolePlayerStarted()) |
| .booleanValue(); |
| } |
| |
| @Override |
| public String description() { |
| return null; |
| } |
| }; |
| await().untilAsserted(ev); |
| |
| System.out.println( |
| "Starting the test and creating the cache and regions etc ..." + System.getenv("PWD")); |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(CACHE_XML_FILE, xmlFilePath); |
| props.put(MAX_WAIT_TIME_RECONNECT, "3000"); |
| props.put(MAX_NUM_RECONNECT_TRIES, "8"); |
| props.put(LOG_LEVEL, "info"); |
| |
| getSystem(props); |
| IgnoredException.addIgnoredException("CacheClosedException"); |
| |
| try { |
| getCache(); |
| } catch (CancelException e) { |
| // can happen if RoleA goes away during initialization |
| System.out.println("cache threw CancelException while creating the cache"); |
| } |
| |
| initialized = true; |
| |
| addReconnectListener(); |
| |
| await().until(() -> reconnectTries != 0); |
| |
| await().until(() -> { |
| String excuse = "none"; |
| if (InternalDistributedSystem.getReconnectAttemptCounter() != 0) { |
| System.out.println("reconnectAttemptCounter is " |
| + InternalDistributedSystem.getReconnectAttemptCounter() |
| + " waiting for it to be zero"); |
| return false; |
| } |
| Object key = null; |
| Object value = null; |
| Region.Entry keyValue = null; |
| try { |
| if (cache == null) { |
| excuse = "no cache"; |
| return false; |
| } |
| Region myRegion = cache.getRegion(regionName); |
| if (myRegion == null) { |
| excuse = "no region"; |
| return false; |
| } |
| |
| Set keyValuePair = myRegion.entrySet(); |
| Iterator it = keyValuePair.iterator(); |
| while (it.hasNext()) { |
| keyValue = (Region.Entry) it.next(); |
| key = keyValue.getKey(); |
| value = keyValue.getValue(); |
| } |
| if (key == null) { |
| excuse = "key is null"; |
| return false; |
| } |
| if (!myKey.equals(key)) { |
| excuse = "key is wrong"; |
| return false; |
| } |
| if (value == null) { |
| excuse = "value is null"; |
| return false; |
| } |
| if (!myValue.equals(value)) { |
| excuse = "value is wrong"; |
| return false; |
| } |
| System.out.println("All assertions passed"); |
| System.out.println("MyKey : " + key + " and myvalue : " + value); |
| return true; |
| } catch (CancelException ecc) { |
| // ignor the exception because the cache can be closed/null some times |
| // while in reconnect. |
| } catch (RegionDestroyedException rex) { |
| |
| } finally { |
| System.out.println("waiting for reconnect. Current status is '" + excuse + "'"); |
| } |
| return false; |
| }); |
| |
| if (cache != null) { |
| cache.getDistributedSystem().disconnect(); |
| } |
| } |
| }; // roleloss runnable |
| } |
| |
| /** |
| * auto-reconnect was found to stop attempting to reconnect and rebuild the cache if another |
| * forced-disconnect was triggered after reconnect but before cache creation was completed. This |
| * test uses a region listener to crash the reconnecting distributed system during cache creation |
| * and asserts that it then reconnects and rebuilds the cache. |
| */ |
| @Test |
| public void testReconnectFailsInCacheCreation() throws Exception { |
| |
| Host host = Host.getHost(0); |
| VM vm0 = host.getVM(0); |
| VM vm1 = host.getVM(1); |
| |
| final int locPort = locatorPort; |
| |
| SerializableRunnable createCache = new SerializableRunnable("Create Cache and Regions") { |
| @Override |
| public void run() { |
| locatorPort = locPort; |
| final Properties props = getDistributedSystemProperties(); |
| props.put(MAX_WAIT_TIME_RECONNECT, "1000"); |
| dsProperties = props; |
| ReconnectDUnitTest.savedSystem = getSystem(props); |
| ReconnectDUnitTest.savedCache = (GemFireCacheImpl) getCache(); |
| Region myRegion = createRegion("myRegion", createAtts()); |
| myRegion.put("MyKey", "MyValue"); |
| myRegion.getAttributesMutator() |
| .addCacheListener(new CacheListenerTriggeringForcedDisconnect()); |
| } |
| }; |
| |
| vm0.invoke(createCache); // vm0 keeps the locator from losing quorum when vm1 crashes |
| |
| vm1.invoke(createCache); |
| IgnoredException.addIgnoredException( |
| "DistributedSystemDisconnectedException|ForcedDisconnectException", vm1); |
| forceDisconnect(vm1); |
| |
| vm1.invoke(new SerializableRunnable("wait for reconnect") { |
| @Override |
| public void run() { |
| final GemFireCacheImpl cache = savedCache; |
| GeodeAwaitility.await().untilAsserted(new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return cache.isReconnecting(); |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for cache to begin reconnecting"; |
| } |
| }); |
| out.println("entering reconnect wait for " + cache); |
| try { |
| cache.waitUntilReconnected(5, MINUTES); |
| } catch (InterruptedException e) { |
| fail("interrupted"); |
| } |
| assertNotNull(cache.getReconnectedCache()); |
| } |
| }); |
| } |
| |
| /** |
| * GEODE-2155 Auto-reconnect fails with NPE due to a cache listener Declarable.init method |
| * throwing an exception. |
| */ |
| @Test |
| public void testReconnectFailsDueToBadCacheXML() throws Exception { |
| |
| Host host = getHost(0); |
| VM vm0 = host.getVM(0); |
| VM vm1 = host.getVM(1); |
| |
| final int locPort = locatorPort; |
| |
| SerializableRunnable createCache = new SerializableRunnable("Create Cache and Regions") { |
| @Override |
| public void run() { |
| locatorPort = locPort; |
| final Properties props = getDistributedSystemProperties(); |
| props.put(MAX_WAIT_TIME_RECONNECT, "1000"); |
| dsProperties = props; |
| savedSystem = getSystem(props); |
| savedCache = (GemFireCacheImpl) getCache(); |
| Region myRegion = createRegion("myRegion", createAtts()); |
| myRegion.put("MyKey", "MyValue"); |
| myRegion.getAttributesMutator().addCacheListener(new ListenerWhoseInitMethodAlwaysThrows()); |
| } |
| }; |
| |
| vm0.invoke(createCache); // vm0 keeps the locator from losing quorum when vm1 crashes |
| |
| createCache.run(); |
| addIgnoredException( |
| "DistributedSystemDisconnectedException|ForcedDisconnectException", vm1); |
| forceDisconnect(null); |
| |
| final GemFireCacheImpl cache = savedCache; |
| GeodeAwaitility.await().untilAsserted(new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return cache.isReconnecting() |
| || cache.getInternalDistributedSystem().isReconnectCancelled(); |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for cache to begin reconnecting"; |
| } |
| }); |
| assertThatThrownBy(() -> cache.waitUntilReconnected(getTimeout().getValueInMS(), MILLISECONDS)) |
| .isInstanceOf(CacheClosedException.class) |
| .hasMessageContaining("Cache could not be recreated") |
| .hasCauseExactlyInstanceOf(DistributedSystemDisconnectedException.class); |
| assertTrue(cache.getInternalDistributedSystem().isReconnectCancelled()); |
| assertNull(cache.getReconnectedCache()); |
| } |
| |
| private CacheSerializableRunnable getRoleAPlayerRunnable(final int locPort, |
| final String regionName, final String myKey, final String myValue, |
| final String startupMessage) { |
| return new CacheSerializableRunnable("second RoleA player") { |
| @Override |
| public void run2() throws CacheException { |
| System.out.println(startupMessage); |
| // closeCache(); |
| // getSystem().disconnect(); |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(LOG_LEVEL, "info"); |
| props.put(ROLES, "RoleA"); |
| |
| getSystem(props); |
| getCache(); |
| AttributesFactory fac = new AttributesFactory(); |
| fac.setScope(Scope.DISTRIBUTED_ACK); |
| fac.setDataPolicy(DataPolicy.REPLICATE); |
| |
| RegionAttributes attr = fac.create(); |
| Region region = createRootRegion(regionName, attr); |
| System.out.println("STARTED THE REQUIREDROLES CACHE"); |
| try { |
| Thread.sleep(120); |
| } catch (Exception ee) { |
| fail("interrupted"); |
| } |
| |
| region.put(myKey, myValue); |
| try { |
| Thread.sleep(5000); // why are we sleeping for 5 seconds here? |
| // if it is to give time to avkVm0 to notice us we should have |
| // avkVm0 signal us that it has seen us and then we can exit. |
| } catch (InterruptedException ee) { |
| fail("interrupted"); |
| } |
| System.out.println("RolePlayer is done..."); |
| |
| |
| } |
| |
| |
| }; |
| } |
| |
| |
| private CacheSerializableRunnable getRoleAPlayerForCacheInitializationRunnable(final VM otherVM, |
| final int locPort, final String regionName, final String startupMessage) { |
| return new CacheSerializableRunnable("first RoleA player") { |
| @Override |
| public void run2() throws CacheException { |
| // closeCache(); |
| // getSystem().disconnect(); |
| System.out.println(startupMessage); |
| locatorPort = locPort; |
| Properties props = getDistributedSystemProperties(); |
| props.put(LOG_LEVEL, "info"); |
| props.put(ROLES, "RoleA"); |
| |
| getSystem(props); |
| getCache(); |
| AttributesFactory fac = new AttributesFactory(); |
| fac.setScope(Scope.DISTRIBUTED_ACK); |
| fac.setDataPolicy(DataPolicy.REPLICATE); |
| |
| RegionAttributes attr = fac.create(); |
| createRootRegion(regionName, attr); |
| System.out.println("STARTED THE REQUIREDROLES CACHE"); |
| initialRolePlayerStarted = true; |
| |
| while (!otherVM.invoke(() -> ReconnectDUnitTest.isInitialized()).booleanValue()) { |
| try { |
| Thread.sleep(15); |
| } catch (InterruptedException ignor) { |
| fail("interrupted"); |
| } |
| } |
| System.out.println("RoleAPlayerInitializer is done..."); |
| closeCache(); |
| |
| } |
| }; |
| } |
| |
| |
| void addReconnectListener() { |
| reconnectTries = 0; // reset the count for this listener |
| System.out.println("adding reconnect listener"); |
| ReconnectListener reconlis = new ReconnectListener() { |
| @Override |
| public void reconnecting(InternalDistributedSystem oldSys) { |
| System.out.println("reconnect listener invoked"); |
| reconnectTries++; |
| } |
| |
| @Override |
| public void onReconnect(InternalDistributedSystem system1, |
| InternalDistributedSystem system2) { |
| System.out.println("reconnect listener onReconnect invoked " + system2); |
| cache = system2.getCache(); |
| } |
| }; |
| InternalDistributedSystem.addReconnectListener(reconlis); |
| } |
| |
| private void waitTimeout() throws InterruptedException { |
| Thread.sleep(500); |
| |
| } |
| |
| public boolean forceDisconnect(VM vm) throws Exception { |
| SerializableCallable fd = new SerializableCallable("crash distributed system") { |
| @Override |
| public Object call() throws Exception { |
| // since the system will disconnect and attempt to reconnect |
| // a new system the old reference to DTC.system can cause |
| // trouble, so we first null it out. |
| nullSystem(); |
| final Locator oldLocator = Locator.getLocator(); |
| final DistributedSystem msys = cache.getDistributedSystem(); |
| MembershipManagerHelper.crashDistributedSystem(msys); |
| if (oldLocator != null) { |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return msys.isReconnecting() || msys.getReconnectedSystem() != null; |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for locator to start reconnecting: " + oldLocator; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); |
| } |
| return true; |
| } |
| }; |
| if (vm != null) { |
| return (Boolean) vm.invoke(fd); |
| } else { |
| return (Boolean) fd.call(); |
| } |
| } |
| |
| /** |
| * A listener whose init always throws an exception. |
| * Since init is always called during cache.xml parsing |
| * this listener is not able to be used from cache.xml. |
| */ |
| public static class ListenerWhoseInitMethodAlwaysThrows extends CacheListenerAdapter { |
| @Override |
| public void init(Properties props) { |
| throw new RuntimeException("Cause parsing to fail"); |
| }; |
| } |
| |
| /** |
| * CacheListenerTriggeringForcedDisconnect crashes the distributed system when it is invoked for |
| * the first time. |
| * After that it ignores any notifications. |
| */ |
| public static class CacheListenerTriggeringForcedDisconnect extends CacheListenerAdapter |
| implements Declarable { |
| public static int crashCount = 0; |
| |
| @Override |
| public void afterRegionCreate(final RegionEvent event) { |
| if (crashCount == 0) { |
| crashCount += 1; |
| // we crash the system in a different thread than the ReconnectThread |
| // to simulate receiving a ForcedDisconnect from the membership manager |
| // in the UDP reader thread |
| Thread t = new Thread("crash reconnecting system (ReconnectDUnitTest)") { |
| @Override |
| public void run() { |
| System.out.println("crashing distributed system"); |
| GemFireCacheImpl cache = (GemFireCacheImpl) event.getRegion().getCache(); |
| MembershipManagerHelper.crashDistributedSystem(cache.getDistributedSystem()); |
| } |
| }; |
| t.setDaemon(true); |
| t.start(); |
| } |
| } |
| |
| @Override |
| public void init(Properties props) {} |
| |
| } |
| |
| protected void addDSProps(Properties p) {} |
| } |
