geode-core/src/distributedTest/java/org/apache/geode/cache30/ReconnectWithClusterConfigurationDUnitTest.java - geode - Git at Google

 /*
  * 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 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.HTTP_SERVICE_PORT;
 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_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.NAME;
 import static org.apache.geode.distributed.ConfigurationProperties.USE_CLUSTER_CONFIGURATION;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
 import java.net.UnknownHostException;
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;

 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;

 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.distributed.DistributedSystem;
 import org.apache.geode.distributed.Locator;
 import org.apache.geode.distributed.internal.InternalLocator;
 import org.apache.geode.distributed.internal.membership.api.MembershipManagerHelper;
 import org.apache.geode.distributed.internal.tcpserver.HostAddress;
 import org.apache.geode.internal.AvailablePortHelper;
 import org.apache.geode.internal.inet.LocalHostUtil;
 import org.apache.geode.internal.membership.utils.AvailablePort;
 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.Disconnect;
 import org.apache.geode.test.dunit.IgnoredException;
 import org.apache.geode.test.dunit.Invoke;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.rules.DistributedRule;

 public class ReconnectWithClusterConfigurationDUnitTest implements Serializable {
   static final int NUM_LOCATORS = 2;
   static final int NUM_VMS = 4;
   static DistributedSystem system;
   static Cache cache;
   static Locator locator;
   static int[] locatorPorts = new int[NUM_LOCATORS];
   static Properties dsProperties;

   @Rule
   public transient DistributedRule distributedRule =
       DistributedRule.builder().withVMCount(NUM_VMS).build();

   @Rule
   public transient TemporaryFolder temporaryFolder = new TemporaryFolder();

   @Before
   public void setup() throws IOException {
     List<AvailablePort.Keeper> randomAvailableTCPPortKeepers =
         AvailablePortHelper.getRandomAvailableTCPPortKeepers(NUM_LOCATORS);
     for (int i = 0; i < NUM_LOCATORS; i++) {
       AvailablePort.Keeper keeper = randomAvailableTCPPortKeepers.get(i);
       locatorPorts[i] = keeper.getPort();
     }
     final int[] locPorts = locatorPorts;
     Invoke.invokeInEveryVM("set locator ports", () -> locatorPorts = locPorts);
     for (int i = 0; i < NUM_LOCATORS; i++) {
       final String workingDir = temporaryFolder.newFolder().getAbsolutePath();
       final int locatorNumber = i;
       randomAvailableTCPPortKeepers.get(locatorNumber).release();
       VM.getVM(i).invoke("start locator", () -> {
         try {
           Disconnect.disconnectFromDS();
           dsProperties = null;
           Properties props = getDistributedSystemProperties();
           locator = InternalLocator.startLocator(locatorPorts[locatorNumber], new File(""),
               null, null, new HostAddress(LocalHostUtil.getLocalHost()), true,
               props, null, Paths.get(workingDir));
           system = locator.getDistributedSystem();
           cache = ((InternalLocator) locator).getCache();
           IgnoredException.addIgnoredException(
               "org.apache.geode.ForcedDisconnectException||Possible loss of quorum");
         } catch (IOException e) {
           Assert.fail("unable to start locator", e);
         }
       });
     }
   }

   @After
   public void teardown() {
     for (int i = 0; i < NUM_LOCATORS; i++) {
       VM.getVM(i).invoke(() -> {
         InternalLocator locator = InternalLocator.getLocator();
         if (locator != null) {
           if (cache != null && cache.isReconnecting()) {
             cache.stopReconnecting();
           }
           locator.stop();
         }
       });
     }
     Invoke.invokeInEveryVM(() -> {
       if (system != null) {
         system.disconnect();
       }
       system = null;
       cache = null;
     });
   }

   public Properties getDistributedSystemProperties() throws UnknownHostException {
     dsProperties = new Properties();
     dsProperties.put(MAX_WAIT_TIME_RECONNECT, "" + (5000 * NUM_VMS));
     dsProperties.put(ENABLE_NETWORK_PARTITION_DETECTION, "true");
     dsProperties.put(DISABLE_AUTO_RECONNECT, "false");
     dsProperties.put(ENABLE_CLUSTER_CONFIGURATION, "true");
     dsProperties.put(USE_CLUSTER_CONFIGURATION, "true");
     dsProperties.put(HTTP_SERVICE_PORT, "0");
     StringBuilder stringBuilder = new StringBuilder();
     final String localHostName = LocalHostUtil.getLocalHostName();
     stringBuilder.append(localHostName + "[")
         .append(locatorPorts[0])
         .append(']');
     for (int i = 1; i < NUM_LOCATORS; i++) {
       stringBuilder.append("," + localHostName + "[")
           .append(locatorPorts[i])
           .append(']');
     }
     dsProperties.put(LOCATORS, stringBuilder.toString());
     dsProperties.put(MCAST_PORT, "0");
     dsProperties.put(MEMBER_TIMEOUT, "5000");
     dsProperties.put(LOG_LEVEL, "info");
     int vmNumber = VM.getCurrentVMNum();
     if (vmNumber < NUM_LOCATORS) {
       dsProperties.put(NAME, "loc" + VM.getCurrentVMNum());
     } else {
       dsProperties.put(NAME, "vm" + VM.getCurrentVMNum());
     }
     return dsProperties;
   }


   @Test
   public void testReconnectAfterMeltdown() throws InterruptedException {

     for (int i = NUM_LOCATORS; i < NUM_VMS; i++) {
       VM.getVM(i).invoke("create cache", () -> {
         cache = new CacheFactory(getDistributedSystemProperties()).create();
         system = cache.getDistributedSystem();
       });
     }
     AsyncInvocation<?>[] crashers = new AsyncInvocation[NUM_VMS];
     for (int i = 0; i < NUM_VMS; i++) {
       crashers[i] = VM.getVM(i).invokeAsync("crash",
           () -> MembershipManagerHelper.crashDistributedSystem(system));
     }
     for (AsyncInvocation<?> crasher : crashers) {
       crasher.join();
     }
     AsyncInvocation<?>[] waiters = new AsyncInvocation[NUM_VMS];
     for (int i = NUM_VMS - 1; i >= 0; i--) {
       waiters[i] = VM.getVM(i).invokeAsync("wait for reconnect", () -> {
         system.waitUntilReconnected(GeodeAwaitility.getTimeout().toMillis(),
             TimeUnit.MILLISECONDS);
         system = system.getReconnectedSystem();
         cache = cache.getReconnectedCache();
         await().untilAsserted(() -> assertThat(system.getAllOtherMembers().size())
             .withFailMessage("wrong number of members: " + system.getAllOtherMembers())
             .isEqualTo(NUM_VMS - 1));
       });
     }
     for (AsyncInvocation<?> waiter : waiters) {
       waiter.join();
     }
   }

 }
	/*
	* 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 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.HTTP_SERVICE_PORT;
	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_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.NAME;
	import static org.apache.geode.distributed.ConfigurationProperties.USE_CLUSTER_CONFIGURATION;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.io.File;
	import java.io.IOException;
	import java.io.Serializable;
	import java.net.UnknownHostException;
	import java.nio.file.Paths;
	import java.util.List;
	import java.util.Properties;
	import java.util.concurrent.TimeUnit;

	import org.junit.After;
	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;
	import org.junit.rules.TemporaryFolder;

	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.CacheFactory;
	import org.apache.geode.distributed.DistributedSystem;
	import org.apache.geode.distributed.Locator;
	import org.apache.geode.distributed.internal.InternalLocator;
	import org.apache.geode.distributed.internal.membership.api.MembershipManagerHelper;
	import org.apache.geode.distributed.internal.tcpserver.HostAddress;
	import org.apache.geode.internal.AvailablePortHelper;
	import org.apache.geode.internal.inet.LocalHostUtil;
	import org.apache.geode.internal.membership.utils.AvailablePort;
	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.Disconnect;
	import org.apache.geode.test.dunit.IgnoredException;
	import org.apache.geode.test.dunit.Invoke;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.rules.DistributedRule;

	public class ReconnectWithClusterConfigurationDUnitTest implements Serializable {
	static final int NUM_LOCATORS = 2;
	static final int NUM_VMS = 4;
	static DistributedSystem system;
	static Cache cache;
	static Locator locator;
	static int[] locatorPorts = new int[NUM_LOCATORS];
	static Properties dsProperties;

	@Rule
	public transient DistributedRule distributedRule =
	DistributedRule.builder().withVMCount(NUM_VMS).build();

	@Rule
	public transient TemporaryFolder temporaryFolder = new TemporaryFolder();

	@Before
	public void setup() throws IOException {
	List<AvailablePort.Keeper> randomAvailableTCPPortKeepers =
	AvailablePortHelper.getRandomAvailableTCPPortKeepers(NUM_LOCATORS);
	for (int i = 0; i < NUM_LOCATORS; i++) {
	AvailablePort.Keeper keeper = randomAvailableTCPPortKeepers.get(i);
	locatorPorts[i] = keeper.getPort();
	}
	final int[] locPorts = locatorPorts;
	Invoke.invokeInEveryVM("set locator ports", () -> locatorPorts = locPorts);
	for (int i = 0; i < NUM_LOCATORS; i++) {
	final String workingDir = temporaryFolder.newFolder().getAbsolutePath();
	final int locatorNumber = i;
	randomAvailableTCPPortKeepers.get(locatorNumber).release();
	VM.getVM(i).invoke("start locator", () -> {
	try {
	Disconnect.disconnectFromDS();
	dsProperties = null;
	Properties props = getDistributedSystemProperties();
	locator = InternalLocator.startLocator(locatorPorts[locatorNumber], new File(""),
	null, null, new HostAddress(LocalHostUtil.getLocalHost()), true,
	props, null, Paths.get(workingDir));
	system = locator.getDistributedSystem();
	cache = ((InternalLocator) locator).getCache();
	IgnoredException.addIgnoredException(
	"org.apache.geode.ForcedDisconnectException\|\|Possible loss of quorum");
	} catch (IOException e) {
	Assert.fail("unable to start locator", e);
	}
	});
	}
	}

	@After
	public void teardown() {
	for (int i = 0; i < NUM_LOCATORS; i++) {
	VM.getVM(i).invoke(() -> {
	InternalLocator locator = InternalLocator.getLocator();
	if (locator != null) {
	if (cache != null && cache.isReconnecting()) {
	cache.stopReconnecting();
	}
	locator.stop();
	}
	});
	}
	Invoke.invokeInEveryVM(() -> {
	if (system != null) {
	system.disconnect();
	}
	system = null;
	cache = null;
	});
	}

	public Properties getDistributedSystemProperties() throws UnknownHostException {
	dsProperties = new Properties();
	dsProperties.put(MAX_WAIT_TIME_RECONNECT, "" + (5000 * NUM_VMS));
	dsProperties.put(ENABLE_NETWORK_PARTITION_DETECTION, "true");
	dsProperties.put(DISABLE_AUTO_RECONNECT, "false");
	dsProperties.put(ENABLE_CLUSTER_CONFIGURATION, "true");
	dsProperties.put(USE_CLUSTER_CONFIGURATION, "true");
	dsProperties.put(HTTP_SERVICE_PORT, "0");
	StringBuilder stringBuilder = new StringBuilder();
	final String localHostName = LocalHostUtil.getLocalHostName();
	stringBuilder.append(localHostName + "[")
	.append(locatorPorts[0])
	.append(']');
	for (int i = 1; i < NUM_LOCATORS; i++) {
	stringBuilder.append("," + localHostName + "[")
	.append(locatorPorts[i])
	.append(']');
	}
	dsProperties.put(LOCATORS, stringBuilder.toString());
	dsProperties.put(MCAST_PORT, "0");
	dsProperties.put(MEMBER_TIMEOUT, "5000");
	dsProperties.put(LOG_LEVEL, "info");
	int vmNumber = VM.getCurrentVMNum();
	if (vmNumber < NUM_LOCATORS) {
	dsProperties.put(NAME, "loc" + VM.getCurrentVMNum());
	} else {
	dsProperties.put(NAME, "vm" + VM.getCurrentVMNum());
	}
	return dsProperties;
	}


	@Test
	public void testReconnectAfterMeltdown() throws InterruptedException {

	for (int i = NUM_LOCATORS; i < NUM_VMS; i++) {
	VM.getVM(i).invoke("create cache", () -> {
	cache = new CacheFactory(getDistributedSystemProperties()).create();
	system = cache.getDistributedSystem();
	});
	}
	AsyncInvocation<?>[] crashers = new AsyncInvocation[NUM_VMS];
	for (int i = 0; i < NUM_VMS; i++) {
	crashers[i] = VM.getVM(i).invokeAsync("crash",
	() -> MembershipManagerHelper.crashDistributedSystem(system));
	}
	for (AsyncInvocation<?> crasher : crashers) {
	crasher.join();
	}
	AsyncInvocation<?>[] waiters = new AsyncInvocation[NUM_VMS];
	for (int i = NUM_VMS - 1; i >= 0; i--) {
	waiters[i] = VM.getVM(i).invokeAsync("wait for reconnect", () -> {
	system.waitUntilReconnected(GeodeAwaitility.getTimeout().toMillis(),
	TimeUnit.MILLISECONDS);
	system = system.getReconnectedSystem();
	cache = cache.getReconnectedCache();
	await().untilAsserted(() -> assertThat(system.getAllOtherMembers().size())
	.withFailMessage("wrong number of members: " + system.getAllOtherMembers())
	.isEqualTo(NUM_VMS - 1));
	});
	}
	for (AsyncInvocation<?> waiter : waiters) {
	waiter.join();
	}
	}

	}