You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2018/10/25 18:41:44 UTC
[geode] branch feature/GEODE-5925 created (now df185b1)
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a change to branch feature/GEODE-5925
in repository https://gitbox.apache.org/repos/asf/geode.git.
at df185b1 GEODE-5925 Server shutdown delays election of new primary bucket owners
This branch includes the following new commits:
new df185b1 GEODE-5925 Server shutdown delays election of new primary bucket owners
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[geode] 01/01: GEODE-5925 Server shutdown delays election of new
primary bucket owners
Posted by bs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch feature/GEODE-5925
in repository https://gitbox.apache.org/repos/asf/geode.git
commit df185b1a82fa45f19cf892a343203f6b9bbc6566
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Thu Oct 25 11:38:03 2018 -0700
GEODE-5925 Server shutdown delays election of new primary bucket owners
This PR addresses the problem of AcceptorImpl sending out expensive
profile update messages when it's being stopped during Cache.close().
There is no reason to send these messages if the affected regions are
going to be destroyed since that also sends profile updates
to other members of the cluster.
---
.../cache/tier/sockets/AcceptorImplJUnitTest.java | 156 +++++++++++----------
.../geode/internal/cache/CacheServerImpl.java | 2 +-
.../AllBucketProfilesUpdateMessage.java | 9 +-
.../internal/cache/tier/sockets/AcceptorImpl.java | 65 +++++----
4 files changed, 119 insertions(+), 113 deletions(-)
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
index 815725b..b4a39e0 100644
--- a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
+++ b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
@@ -15,9 +15,10 @@
package org.apache.geode.internal.cache.tier.sockets;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.net.BindException;
@@ -26,9 +27,9 @@ import java.util.Properties;
import org.junit.After;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.mockito.Mockito;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
@@ -36,7 +37,6 @@ import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
-import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.test.junit.categories.ClientServerTest;
@@ -45,6 +45,8 @@ public class AcceptorImplJUnitTest {
DistributedSystem system;
InternalCache cache;
+ AcceptorImpl a1 = null, a2 = null;
+ ServerConnectionFactory serverConnectionFactory = new ServerConnectionFactory();
@Before
public void setUp() throws Exception {
@@ -56,85 +58,87 @@ public class AcceptorImplJUnitTest {
@After
public void tearDown() throws Exception {
+ if (a1 != null) {
+ a1.close();
+ }
+ if (a2 != null) {
+ a2.close();
+ }
this.cache.close();
this.system.disconnect();
}
- /*
- * Test method for 'org.apache.geode.internal.cache.tier.sockets.AcceptorImpl(int, int, boolean,
- * int, Cache)'
- */
- @Ignore
- @Test
- public void testConstructor() throws CacheException, IOException {
- AcceptorImpl a1 = null, a2 = null, a3 = null;
- try {
- final int[] freeTCPPorts = AvailablePortHelper.getRandomAvailableTCPPorts(2);
- int port1 = freeTCPPorts[0];
- int port2 = freeTCPPorts[1];
-
-
- ServerConnectionFactory serverConnectionFactory = new ServerConnectionFactory();
- try {
- new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
- AcceptorImpl.MINIMUM_MAX_CONNECTIONS - 1, CacheServer.DEFAULT_MAX_THREADS,
- CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE,
- null, null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
- serverConnectionFactory);
- fail("Expected an IllegalArgumentExcption due to max conns < min pool size");
- } catch (IllegalArgumentException expected) {
- }
+ @Test(expected = BindException.class)
+ public void constructorThrowsBindException() throws CacheException, IOException {
+ a1 = new AcceptorImpl(0, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
+ CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE,
+ null, null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
+ serverConnectionFactory, 1000);
+ a2 = new AcceptorImpl(a1.getPort(), null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
+ CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE,
+ null, null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
+ serverConnectionFactory, 1000);
+ }
- try {
- new AcceptorImpl(port2, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache, 0,
- CacheServer.DEFAULT_MAX_THREADS, CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT,
- CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE, null, null, false, Collections.EMPTY_LIST,
- CacheServer.DEFAULT_TCP_NO_DELAY, serverConnectionFactory);
- fail("Expected an IllegalArgumentExcption due to max conns of zero");
- } catch (IllegalArgumentException expected) {
- }
+ @Test
+ public void acceptorBindsToLocalAddress() throws CacheException, IOException {
+ a1 = new AcceptorImpl(0, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
+ CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE, null,
+ null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
+ serverConnectionFactory, 1000);
+ InternalDistributedSystem isystem =
+ (InternalDistributedSystem) this.cache.getDistributedSystem();
+ DistributionConfig config = isystem.getConfig();
+ String bindAddress = config.getBindAddress();
+ if (bindAddress == null || bindAddress.length() <= 0) {
+ assertTrue(a1.getServerInetAddr().isAnyLocalAddress());
+ }
+ }
- try {
- a1 = new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
- AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
- CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE,
- null, null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
- serverConnectionFactory);
- a2 = new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
- AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
- CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE,
- null, null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
- serverConnectionFactory);
- fail("Expecetd a BindException while attaching to the same port");
- } catch (BindException expected) {
- }
+ /**
+ * If a CacheServer is stopped but the cache is still open we need to inform other members
+ * of the cluster that the server component no longer exists. Partitioned Region bucket
+ * advisors need to know about this event.
+ */
+ @Test
+ public void acceptorCloseInformsOtherServersIfCacheIsNotClosed() throws Exception {
+ a1 = new AcceptorImpl(0, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
+ CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE, null,
+ null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
+ serverConnectionFactory, 1000);
+ InternalDistributedSystem isystem =
+ (InternalDistributedSystem) this.cache.getDistributedSystem();
+ AcceptorImpl spy = Mockito.spy(a1);
+ spy.close();
+ verify(spy, atLeastOnce()).notifyCacheMembersOfClose();
+ }
- a3 = new AcceptorImpl(port2, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
- AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
- CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE, null,
- null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
- serverConnectionFactory);
- assertEquals(port2, a3.getPort());
- InternalDistributedSystem isystem =
- (InternalDistributedSystem) this.cache.getDistributedSystem();
- DistributionConfig config = isystem.getConfig();
- String bindAddress = config.getBindAddress();
- if (bindAddress == null || bindAddress.length() <= 0) {
- assertTrue(a3.getServerInetAddr().isAnyLocalAddress());
- }
- } finally {
- if (a1 != null)
- a1.close();
- if (a2 != null)
- a2.close();
- if (a3 != null)
- a3.close();
- }
+ /**
+ * If a CacheServer is stopped as part of cache.close() we don't need to inform other
+ * members of the cluster since all regions will be destroyed.
+ */
+ @Test
+ public void acceptorCloseDoesNotInformOtherServersIfCacheIsClosed() throws Exception {
+ a1 = new AcceptorImpl(0, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ AcceptorImpl.MINIMUM_MAX_CONNECTIONS, CacheServer.DEFAULT_MAX_THREADS,
+ CacheServer.DEFAULT_MAXIMUM_MESSAGE_COUNT, CacheServer.DEFAULT_MESSAGE_TIME_TO_LIVE, null,
+ null, false, Collections.EMPTY_LIST, CacheServer.DEFAULT_TCP_NO_DELAY,
+ serverConnectionFactory, 1000);
+ InternalDistributedSystem isystem =
+ (InternalDistributedSystem) this.cache.getDistributedSystem();
+ AcceptorImpl spy = Mockito.spy(a1);
+ cache.close();
+ spy.close();
+ verify(spy, never()).notifyCacheMembersOfClose();
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/CacheServerImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/CacheServerImpl.java
index dc7cdef..517c386 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/CacheServerImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/CacheServerImpl.java
@@ -351,7 +351,7 @@ public class CacheServerImpl extends AbstractCacheServer implements Distribution
getSocketBufferSize(), getMaximumTimeBetweenPings(), this.cache, getMaxConnections(),
getMaxThreads(), getMaximumMessageCount(), getMessageTimeToLive(), this.loadMonitor,
overflowAttributesList, this.isGatewayReceiver, this.gatewayTransportFilters,
- this.tcpNoDelay, serverConnectionFactory);
+ this.tcpNoDelay, serverConnectionFactory, 120000);
this.acceptor.start();
this.advisor.handshake();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/AllBucketProfilesUpdateMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/AllBucketProfilesUpdateMessage.java
index 0d7c32c..3f69ef6 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/AllBucketProfilesUpdateMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/AllBucketProfilesUpdateMessage.java
@@ -113,21 +113,18 @@ public class AllBucketProfilesUpdateMessage extends DistributionMessage
* @param dm the distribution manager used to send the message
* @param prId the unique partitioned region identifier
* @param profiles bucked id to profile map
- * @param requireAck whether or not to expect a reply
* @return an instance of reply processor if requireAck is true on which the caller can wait until
* the event has finished.
*/
public static ReplyProcessor21 send(Set recipients, DistributionManager dm, int prId,
- Map<Integer, BucketAdvisor.BucketProfile> profiles, boolean requireAck) {
+ Map<Integer, BucketAdvisor.BucketProfile> profiles) {
if (recipients.isEmpty()) {
return null;
}
ReplyProcessor21 rp = null;
int procId = 0;
- if (requireAck) {
- rp = new ReplyProcessor21(dm, recipients);
- procId = rp.getProcessorId();
- }
+ rp = new ReplyProcessor21(dm, recipients);
+ procId = rp.getProcessorId();
AllBucketProfilesUpdateMessage m =
new AllBucketProfilesUpdateMessage(recipients, prId, procId, profiles);
dm.putOutgoing(m);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
index 85c10d2..6de608e 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
@@ -334,11 +334,14 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
* @since GemFire 5.7
*/
public AcceptorImpl(int port, String bindHostName, boolean notifyBySubscription,
- int socketBufferSize, int maximumTimeBetweenPings, InternalCache internalCache,
- int maxConnections, int maxThreads, int maximumMessageCount, int messageTimeToLive,
- ConnectionListener listener, List overflowAttributesList, boolean isGatewayReceiver,
+ int socketBufferSize, int maximumTimeBetweenPings,
+ InternalCache internalCache,
+ int maxConnections, int maxThreads, int maximumMessageCount,
+ int messageTimeToLive,
+ ConnectionListener listener, List overflowAttributesList,
+ boolean isGatewayReceiver,
List<GatewayTransportFilter> transportFilter, boolean tcpNoDelay,
- ServerConnectionFactory serverConnectionFactory) throws IOException {
+ ServerConnectionFactory serverConnectionFactory, long timeLimitMillis) throws IOException {
this.securityService = internalCache.getSecurityService();
this.bindHostName = calcBindHostName(internalCache, bindHostName);
this.connectionListener = listener == null ? new ConnectionListenerAdapter() : listener;
@@ -430,7 +433,7 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
gc = null;
}
final int backLog = Integer.getInteger(BACKLOG_PROPERTY_NAME, DEFAULT_BACKLOG).intValue();
- final long tilt = System.currentTimeMillis() + 120 * 1000;
+ final long tilt = System.currentTimeMillis() + timeLimitMillis;
if (isSelector()) {
if (this.socketCreator.useSSL()) {
@@ -692,7 +695,7 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
receipients = pr.getRegionAdvisor().adviseAllPRNodes();
// send it to all in one messgae
ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(receipients,
- pr.getDistributionManager(), pr.getPRId(), profiles, true);
+ pr.getDistributionManager(), pr.getPRId(), profiles);
if (reply != null) {
reply.waitForRepliesUninterruptibly();
}
@@ -1550,38 +1553,40 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
this.clientNotifier.shutdown(this.acceptorId);
shutdownPools();
this.stats.close();
- notifyCacheMembersOfClose();
+ if (!cache.isClosed()) {
+ // the cache isn't closing so we need to inform peers that this CacheServer no longer
+ // exists
+ notifyCacheMembersOfClose();
+ }
} // synchronized
} catch (RuntimeException e) {/* ignore and log */
logger.warn("unexpected", e);
}
}
- private void notifyCacheMembersOfClose() {
- if (!this.cache.forcedDisconnect()) {
- for (PartitionedRegion pr : this.cache.getPartitionedRegions()) {
- Map<Integer, BucketAdvisor.BucketProfile> profiles = new HashMap<>();
- // get all local real bucket advisors
- Map<Integer, BucketAdvisor> advisors = pr.getRegionAdvisor().getAllBucketAdvisors();
- for (Map.Entry<Integer, BucketAdvisor> entry : advisors.entrySet()) {
- BucketAdvisor advisor = entry.getValue();
- BucketProfile bp = (BucketProfile) advisor.createProfile();
- advisor.updateServerBucketProfile(bp);
- profiles.put(entry.getKey(), bp);
- }
-
- Set recipients = pr.getRegionAdvisor().adviseAllPRNodes();
- // send it to all in one message
- ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(recipients,
- pr.getDistributionManager(), pr.getPRId(), profiles, true);
- if (reply != null) {
- reply.waitForRepliesUninterruptibly();
- }
+ void notifyCacheMembersOfClose() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("sending messages to all peers for removing this server..");
+ }
+ for (PartitionedRegion pr : this.cache.getPartitionedRegions()) {
+ Map<Integer, BucketAdvisor.BucketProfile> profiles = new HashMap<>();
+ // get all local real bucket advisors
+ Map<Integer, BucketAdvisor> advisors = pr.getRegionAdvisor().getAllBucketAdvisors();
+ for (Map.Entry<Integer, BucketAdvisor> entry : advisors.entrySet()) {
+ BucketAdvisor advisor = entry.getValue();
+ BucketProfile bp = (BucketProfile) advisor.createProfile();
+ advisor.updateServerBucketProfile(bp);
+ profiles.put(entry.getKey(), bp);
+ }
- if (logger.isDebugEnabled()) {
- logger.debug("sending messages to all peers for removing this server..");
- }
+ Set recipients = pr.getRegionAdvisor().adviseAllPRNodes();
+ // send it to all in one message
+ ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(recipients,
+ pr.getDistributionManager(), pr.getPRId(), profiles);
+ if (reply != null) {
+ reply.waitForRepliesUninterruptibly();
}
+
}
}