You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by kl...@apache.org on 2017/02/09 23:34:35 UTC
[01/12] geode git commit: [GEODE-2324] Update isShutdownProperly
method. [Forced Update!]
Repository: geode
Updated Branches:
refs/heads/feature/GEODE-2456 d4f8c8068 -> 7085c3205 (forced update)
[GEODE-2324] Update isShutdownProperly method.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/cd229110
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/cd229110
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/cd229110
Branch: refs/heads/feature/GEODE-2456
Commit: cd2291102744f4bb1b8bb5170e7388b31295ef6f
Parents: d73ec97
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Mon Jan 30 11:40:09 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../apache/geode/internal/cache/tier/sockets/AcceptorImpl.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/cd229110/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 5fa8096..ed29472 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
@@ -1634,9 +1634,10 @@ public class AcceptorImpl extends Acceptor implements Runnable {
}
public boolean isShutdownProperly() {
- return !isRunning() && (selectorThread == null || !selectorThread.isAlive())
+ return !isRunning() && !thread.isAlive()
+ && (selectorThread == null || !selectorThread.isAlive())
&& (pool == null || pool.isShutdown()) && (hsPool == null || hsPool.isShutdown())
- && (selector == null || !selector.isOpen());
+ && (selector == null || !selector.isOpen()) && (tmpSel == null || !tmpSel.isOpen());
}
/**
[08/12] geode git commit: [GEODE-2295] Add constructors for
LocatorCancelException
Posted by kl...@apache.org.
[GEODE-2295] Add constructors for LocatorCancelException
and update one place where it's used.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/ec245114
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/ec245114
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/ec245114
Branch: refs/heads/feature/GEODE-2456
Commit: ec2451147154320ccd4a0b0e3b687fd324783e6e
Parents: cd22911
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Wed Jan 11 17:00:05 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:33:32 2017 -0800
----------------------------------------------------------------------
.../internal/tcpserver/LocatorCancelException.java | 13 ++++++++++++-
.../distributed/internal/tcpserver/TcpClient.java | 2 +-
2 files changed, 13 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/ec245114/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
index aec0fe6..1c07a87 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
@@ -17,6 +17,17 @@ package org.apache.geode.distributed.internal.tcpserver;
import org.apache.geode.CancelException;
public class LocatorCancelException extends CancelException {
+ public LocatorCancelException(final String message) {
+ super(message);
+ }
- public LocatorCancelException(final String s) {}
+ public LocatorCancelException(Throwable cause) {
+ super(cause);
+ }
+
+ public LocatorCancelException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public LocatorCancelException() {}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/ec245114/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpClient.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpClient.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpClient.java
index 2e27f3d..47b5703 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpClient.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/TcpClient.java
@@ -265,7 +265,7 @@ public class TcpClient {
sock = socketCreator.connect(ipAddr.getAddress(), ipAddr.getPort(), timeout, null, false);
sock.setSoTimeout(timeout);
} catch (SSLHandshakeException e) {
- throw new LocatorCancelException("Unrecognisable response received");
+ throw new LocatorCancelException("Unrecognisable response received", e);
}
try {
[10/12] geode git commit: GEODE-2455: use spotlessApply to fix format
Posted by kl...@apache.org.
GEODE-2455: use spotlessApply to fix format
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/f7a8ae67
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/f7a8ae67
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/f7a8ae67
Branch: refs/heads/feature/GEODE-2456
Commit: f7a8ae679c731c91dfd6d414e46429a3abaa5754
Parents: a61c55b
Author: Kirk Lund <kl...@apache.org>
Authored: Thu Feb 9 15:12:04 2017 -0800
Committer: Kirk Lund <kl...@apache.org>
Committed: Thu Feb 9 15:12:29 2017 -0800
----------------------------------------------------------------------
.../DescribeDiskStoreFunctionJUnitTest.java | 17 +++++++++--------
1 file changed, 9 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/f7a8ae67/geode-core/src/test/java/org/apache/geode/management/internal/cli/functions/DescribeDiskStoreFunctionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/management/internal/cli/functions/DescribeDiskStoreFunctionJUnitTest.java b/geode-core/src/test/java/org/apache/geode/management/internal/cli/functions/DescribeDiskStoreFunctionJUnitTest.java
index 1dd504a..e0038ef 100644
--- a/geode-core/src/test/java/org/apache/geode/management/internal/cli/functions/DescribeDiskStoreFunctionJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/internal/cli/functions/DescribeDiskStoreFunctionJUnitTest.java
@@ -286,9 +286,9 @@ public class DescribeDiskStoreFunctionJUnitTest {
@Test
public void testAssertStateThrowsIllegalStateException() {
- assertThatThrownBy(() ->
- DescribeDiskStoreFunction.assertState(false, "Expected (%1$s) message!", "test"))
- .isInstanceOf(IllegalStateException.class).hasMessage("Expected (test) message!");
+ assertThatThrownBy(
+ () -> DescribeDiskStoreFunction.assertState(false, "Expected (%1$s) message!", "test"))
+ .isInstanceOf(IllegalStateException.class).hasMessage("Expected (test) message!");
}
private void setupEmptyRegionsPdxGatewaysCacheServersAndAsyncEventQueues(
@@ -644,9 +644,9 @@ public class DescribeDiskStoreFunctionJUnitTest {
function.execute(mockFunctionContext);
String expected = String.format("A disk store with name (%1$s) was not found on member (%2$s).",
- diskStoreName, memberName);
+ diskStoreName, memberName);
assertThatThrownBy(() -> testResultSender.getResults())
- .isInstanceOf(DiskStoreNotFoundException.class).hasMessage(expected);
+ .isInstanceOf(DiskStoreNotFoundException.class).hasMessage(expected);
}
@Test
@@ -686,8 +686,8 @@ public class DescribeDiskStoreFunctionJUnitTest {
function.execute(mockFunctionContext);
- assertThatThrownBy(() -> testResultSender.getResults())
- .isInstanceOf(RuntimeException.class).hasMessage("ExpectedStrings");
+ assertThatThrownBy(() -> testResultSender.getResults()).isInstanceOf(RuntimeException.class)
+ .hasMessage("ExpectedStrings");
}
@Test
@@ -744,7 +744,8 @@ public class DescribeDiskStoreFunctionJUnitTest {
function.execute(mockFunctionContext);
- String expected = "The number of disk directories with a specified size (0) does not match the number of disk directories (1)!";
+ String expected =
+ "The number of disk directories with a specified size (0) does not match the number of disk directories (1)!";
assertThatThrownBy(() -> testResultSender.getResults()).hasMessage(expected);
}
[09/12] geode git commit: [GEODE-2295] Add super() to empty
constructor,
Posted by kl...@apache.org.
[GEODE-2295] Add super() to empty constructor,
remove final to be consistent.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/a61c55b9
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/a61c55b9
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/a61c55b9
Branch: refs/heads/feature/GEODE-2456
Commit: a61c55b9b0d4fffc70a1502f06cbca59098c9418
Parents: ec24511
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Mon Jan 16 11:51:52 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:36:36 2017 -0800
----------------------------------------------------------------------
.../distributed/internal/tcpserver/LocatorCancelException.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/a61c55b9/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
index 1c07a87..7a4ef38 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/tcpserver/LocatorCancelException.java
@@ -17,7 +17,7 @@ package org.apache.geode.distributed.internal.tcpserver;
import org.apache.geode.CancelException;
public class LocatorCancelException extends CancelException {
- public LocatorCancelException(final String message) {
+ public LocatorCancelException(String message) {
super(message);
}
@@ -29,5 +29,7 @@ public class LocatorCancelException extends CancelException {
super(message, cause);
}
- public LocatorCancelException() {}
+ public LocatorCancelException() {
+ super();
+ }
}
[11/12] geode git commit: GEODE-2456: change JUnit4CacheTestCase
tearDown to use deleteQuietly
Posted by kl...@apache.org.
GEODE-2456: change JUnit4CacheTestCase tearDown to use deleteQuietly
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/15748ed4
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/15748ed4
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/15748ed4
Branch: refs/heads/feature/GEODE-2456
Commit: 15748ed41ff6f3512ef1c69d0c64c2fbd42d6707
Parents: f7a8ae6
Author: Kirk Lund <kl...@apache.org>
Authored: Thu Feb 9 12:02:56 2017 -0800
Committer: Kirk Lund <kl...@apache.org>
Committed: Thu Feb 9 15:15:34 2017 -0800
----------------------------------------------------------------------
.../test/dunit/cache/internal/JUnit4CacheTestCase.java | 10 +++-------
1 file changed, 3 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/15748ed4/geode-core/src/test/java/org/apache/geode/test/dunit/cache/internal/JUnit4CacheTestCase.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/cache/internal/JUnit4CacheTestCase.java b/geode-core/src/test/java/org/apache/geode/test/dunit/cache/internal/JUnit4CacheTestCase.java
index a511995..f8b1415 100644
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/cache/internal/JUnit4CacheTestCase.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/cache/internal/JUnit4CacheTestCase.java
@@ -22,6 +22,7 @@ import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
+import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.Logger;
import org.apache.geode.cache.AttributesFactory;
@@ -41,7 +42,6 @@ import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
-import org.apache.geode.internal.FileUtil;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.LocalRegion;
@@ -568,17 +568,13 @@ public abstract class JUnit4CacheTestCase extends JUnit4DistributedTestCase
}
public static final void cleanDiskDirs() throws IOException {
- FileUtil.delete(getDiskDir());
+ FileUtils.deleteQuietly(getDiskDir());
Arrays.stream(new File(".").listFiles()).forEach(file -> deleteBACKUPDiskStoreFile(file));
}
private static void deleteBACKUPDiskStoreFile(final File file) {
if (file.getName().startsWith("BACKUPDiskStore-")) {
- try {
- FileUtil.delete(file);
- } catch (IOException e) {
- throw new RuntimeException("Unable to delete BACKUPDiskStore file", e);
- }
+ FileUtils.deleteQuietly(file);
}
}
[07/12] geode git commit: [GEODE-2324] Dunit test for
AcceptorImpl.close()
Posted by kl...@apache.org.
[GEODE-2324] Dunit test for AcceptorImpl.close()
* Add AcceptorImplObserver
* Add AccceptorImplDunitTest
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/1d561dcc
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/1d561dcc
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/1d561dcc
Branch: refs/heads/feature/GEODE-2456
Commit: 1d561dccaf8736972c6d355244b0fe54b9bd7a8b
Parents: cae580f
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Mon Jan 23 21:49:35 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../cache/tier/sockets/AcceptorImpl.java | 12 ++
.../sockets/command/AcceptorImplObserver.java | 47 ++++++
.../tier/sockets/AcceptorImplDUnitTest.java | 164 +++++++++++++++++++
.../tier/sockets/AcceptorImplJUnitTest.java | 42 ++---
4 files changed, 239 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/1d561dcc/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 37438b6..d72c72b 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
@@ -57,6 +57,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.net.ssl.SSLException;
+import org.apache.geode.internal.cache.tier.sockets.command.AcceptorImplObserver;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
@@ -1528,8 +1529,12 @@ public class AcceptorImpl extends Acceptor implements Runnable {
@Override
public void close() {
+ AcceptorImplObserver acceptorImplObserver = AcceptorImplObserver.getInstance();
try {
synchronized (syncLock) {
+ if (acceptorImplObserver != null) {
+ acceptorImplObserver.beforeClose(this);
+ }
if (!isRunning()) {
return;
}
@@ -1608,9 +1613,16 @@ public class AcceptorImpl extends Acceptor implements Runnable {
}
}
}
+ if (acceptorImplObserver != null) {
+ acceptorImplObserver.normalCloseTermination(this);
+ }
} // synchronized
} catch (RuntimeException e) {/* ignore and log */
logger.warn(LocalizedMessage.create(LocalizedStrings.AcceptorImpl_UNEXPECTED), e);
+ } finally {
+ if (acceptorImplObserver != null) {
+ acceptorImplObserver.afterClose(this);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/1d561dcc/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
new file mode 100644
index 0000000..f5ee982
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
@@ -0,0 +1,47 @@
+/*
+ * 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.tier.sockets.command;
+
+import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
+
+/**
+ * AcceptorImplObserver is an observer/visitor for AcceptorImpl that is used for testing.
+ */
+public abstract class AcceptorImplObserver {
+ private static AcceptorImplObserver instance;
+
+ /**
+ * Set the instance of the observer. Setting to null will clear the observer.
+ *
+ * @param instance
+ * @return the old observer, or null if there was no old observer.
+ */
+ public static final AcceptorImplObserver setInstance(AcceptorImplObserver instance) {
+ AcceptorImplObserver oldInstance = AcceptorImplObserver.instance;
+ AcceptorImplObserver.instance = instance;
+ return oldInstance;
+ }
+
+ public static final AcceptorImplObserver getInstance() {
+ return instance;
+ }
+
+ public void beforeClose(AcceptorImpl acceptorImpl) {}
+
+ public void normalCloseTermination(AcceptorImpl acceptorImpl) {}
+
+ public void afterClose(AcceptorImpl acceptorImpl) {}
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/1d561dcc/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
new file mode 100644
index 0000000..8b4c672
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.tier.sockets;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.cache.client.ClientRegionShortcut;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.cache.util.CacheWriterAdapter;
+import org.apache.geode.internal.AvailablePortHelper;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.sockets.command.AcceptorImplObserver;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests for AcceptorImpl.
+ */
+@Category(DistributedTest.class)
+public class AcceptorImplDUnitTest extends JUnit4DistributedTestCase {
+ private static Cache cache;
+
+ public AcceptorImplDUnitTest() {
+ super();
+ }
+
+ @Override
+ public void postTearDown() throws Exception {
+ if (cache != null) {
+ cache.close();
+ cache = null;
+ }
+ super.postTearDown();
+ }
+
+ public static class SleepyCacheWriter<K, V> extends CacheWriterAdapter<K, V> {
+ @Override
+ public void beforeCreate(EntryEvent<K, V> event) {
+ while (true) {
+ System.out.println("Sleeping a long time.");
+ try {
+ Thread.sleep(100000000);
+ } catch (InterruptedException ignore) {
+ }
+ }
+ }
+ }
+
+ /**
+ * Dump threads to standard out. For debugging.
+ */
+ private void dumpThreads() {
+ ThreadMXBean bean = ManagementFactory.getThreadMXBean();
+ ThreadInfo[] infos = bean.dumpAllThreads(true, true);
+ System.out.println("infos = " + Arrays.toString(infos));
+ }
+
+ /**
+ * GEODE-2324. There was a bug where, due to an uncaught exception, `AcceptorImpl.close()` was
+ * short-circuiting and failing to clean up properly.
+ *
+ * What this test does is start a Cache and hook the Acceptor to interrupt the thread before the
+ * place where an InterruptedException could be thrown. It interrupts the thread, and checks that
+ * the thread has terminated normally without short-circuiting. It doesn't check that every part
+ * of the AcceptorImpl has shut down properly -- that seems both difficult to check (especially
+ * since the fields are private) and implementation-dependent.
+ */
+ @Test
+ public void testShutdownCatchesException() throws Exception {
+ final String hostname = Host.getHost(0).getHostName();
+ final VM clientVM = Host.getHost(0).getVM(0);
+
+ // AtomicBooleans can be set from wherever they are, including an anonymous class or other
+ // thread.
+ AtomicBoolean terminatedNormally = new AtomicBoolean(false);
+ AtomicBoolean passedPostConditions = new AtomicBoolean(false);
+
+ Properties props = new Properties();
+ props.setProperty(MCAST_PORT, "0");
+
+ AcceptorImplObserver.setInstance(new AcceptorImplObserver() {
+ @Override
+ public void beforeClose(AcceptorImpl acceptorImpl) {
+ Thread.currentThread().interrupt();
+ }
+
+ @Override
+ public void normalCloseTermination(AcceptorImpl acceptorImpl) {
+ terminatedNormally.set(true);
+ }
+
+ @Override
+ public void afterClose(AcceptorImpl acceptorImpl) {
+ passedPostConditions.set(!acceptorImpl.isRunning());
+ }
+ });
+
+ try (InternalCache cache = (InternalCache) new CacheFactory(props).create()) {
+ RegionFactory<Object, Object> regionFactory =
+ cache.createRegionFactory(RegionShortcut.PARTITION);
+
+ regionFactory.setCacheWriter(new SleepyCacheWriter<>());
+
+ final CacheServer server = cache.addCacheServer();
+ final int port = AvailablePortHelper.getRandomAvailableTCPPort();
+ server.setPort(port);
+ server.start();
+
+ regionFactory.create("region1");
+
+ clientVM.invokeAsync(() -> {
+ ClientCacheFactory clientCacheFactory = new ClientCacheFactory();
+ clientCacheFactory.addPoolServer(hostname, port);
+ ClientCache clientCache = clientCacheFactory.create();
+ Region<Object, Object> clientRegion1 =
+ clientCache.createClientRegionFactory(ClientRegionShortcut.PROXY).create("region1");
+ clientRegion1.put("foo", "bar");
+ });
+
+ cache.close();
+
+ dumpThreads();
+ assertTrue(terminatedNormally.get());
+ assertTrue(passedPostConditions.get());
+
+ // cleanup.
+ AcceptorImplObserver.setInstance(null);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/1d561dcc/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
index 7aa11b7..58c2157 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
@@ -15,9 +15,13 @@
package org.apache.geode.internal.cache.tier.sockets;
import org.apache.geode.cache.AttributesFactory;
+import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ServerRefusedConnectionException;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedSystem;
@@ -42,30 +46,16 @@ import java.net.BindException;
import java.net.Socket;
import java.util.Collections;
import java.util.Properties;
+import java.util.Set;
+import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID.system;
import static org.junit.Assert.*;
@Category({IntegrationTest.class, ClientServerTest.class})
public class AcceptorImplJUnitTest {
- DistributedSystem system;
- InternalCache cache;
-
- @Before
- public void setUp() throws Exception {
- Properties p = new Properties();
- p.setProperty(MCAST_PORT, "0");
- this.system = DistributedSystem.connect(p);
- this.cache = (InternalCache) CacheFactory.create(system);
- }
-
- @After
- public void tearDown() throws Exception {
- this.cache.close();
- this.system.disconnect();
- }
-
/*
* Test method for 'org.apache.geode.internal.cache.tier.sockets.AcceptorImpl(int, int, boolean,
* int, Cache)'
@@ -74,15 +64,16 @@ public class AcceptorImplJUnitTest {
@Test
public void testConstructor() throws CacheException, IOException {
AcceptorImpl a1 = null, a2 = null, a3 = null;
- try {
+ Properties props = new Properties();
+ props.setProperty(MCAST_PORT, "0");
+ try (InternalCache cache = (InternalCache) new CacheFactory(props).create()) {
final int[] freeTCPPorts = AvailablePortHelper.getRandomAvailableTCPPorts(2);
int port1 = freeTCPPorts[0];
int port2 = freeTCPPorts[1];
-
try {
new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, 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);
@@ -92,7 +83,7 @@ public class AcceptorImplJUnitTest {
try {
new AcceptorImpl(port2, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache, 0,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, 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);
@@ -102,12 +93,12 @@ public class AcceptorImplJUnitTest {
try {
a1 = new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, 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);
a2 = new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, 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);
@@ -116,13 +107,12 @@ public class AcceptorImplJUnitTest {
}
a3 = new AcceptorImpl(port2, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, this.cache,
+ CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, 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);
assertEquals(port2, a3.getPort());
- InternalDistributedSystem isystem =
- (InternalDistributedSystem) this.cache.getDistributedSystem();
+ InternalDistributedSystem isystem = (InternalDistributedSystem) cache.getDistributedSystem();
DistributionConfig config = isystem.getConfig();
String bindAddress = config.getBindAddress();
if (bindAddress == null || bindAddress.length() <= 0) {
[02/12] geode git commit: [GEODE-2324] keep private variables private.
Posted by kl...@apache.org.
[GEODE-2324] keep private variables private.
* Update AcceptorImplDUnit test,
* Refactor AcceptorImpl.close into multiple methods.
* Remove a commented-out method.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/d73ec978
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/d73ec978
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/d73ec978
Branch: refs/heads/feature/GEODE-2456
Commit: d73ec978476ec4bf835c38d1713e14b48324515f
Parents: 6bed282
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Wed Jan 25 11:30:07 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../cache/tier/sockets/AcceptorImpl.java | 202 +++++++-----------
.../sockets/command/AcceptorImplObserver.java | 29 ---
.../tier/sockets/AcceptorImplDUnitTest.java | 205 +++++++++++++------
.../tier/sockets/AcceptorImplJUnitTest.java | 42 ++--
4 files changed, 243 insertions(+), 235 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/d73ec978/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 1bca1fd..5fa8096 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
@@ -57,7 +57,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.net.ssl.SSLException;
-import org.apache.geode.internal.cache.tier.sockets.command.AcceptorImplObserver;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
@@ -207,7 +206,7 @@ public class AcceptorImpl extends Acceptor implements Runnable {
public final AtomicInteger clientServerCnxCount = new AtomicInteger();
/** Has this acceptor been shut down */
- private volatile boolean shutdown = false;
+ private volatile boolean shutdownStarted = false;
/** The thread that runs the acceptor */
private Thread thread = null;
@@ -271,33 +270,6 @@ public class AcceptorImpl extends Acceptor implements Runnable {
private SecurityService securityService = IntegratedSecurityService.getSecurityService();
- // Assumed non-null. Do not set this to null.
- private static AcceptorImplObserver acceptorImplObserver_do_not_access_directly =
- new AcceptorImplObserver() {
- @Override
- public void beforeClose(AcceptorImpl acceptorImpl) {}
-
- @Override
- public void normalCloseTermination(AcceptorImpl acceptorImpl) {}
-
- @Override
- public void afterClose(AcceptorImpl acceptorImpl) {}
- };
-
- private static AcceptorImplObserver getAcceptorImplObserver() {
- synchronized (AcceptorImpl.class) {
- return acceptorImplObserver_do_not_access_directly;
- }
- }
-
- public static void setObserver_TESTONLY(AcceptorImplObserver observer) {
- synchronized (AcceptorImpl.class) {
- if (observer != null) {
- acceptorImplObserver_do_not_access_directly = observer;
- }
- }
- }
-
/**
* Initializes this acceptor thread to listen for connections on the given port.
*
@@ -1551,19 +1523,17 @@ public class AcceptorImpl extends Acceptor implements Runnable {
@Override
public boolean isRunning() {
- return !this.shutdown;
+ return !this.shutdownStarted;
}
@Override
public void close() {
- AcceptorImplObserver acceptorImplObserver = getAcceptorImplObserver();
try {
synchronized (syncLock) {
- acceptorImplObserver.beforeClose(this);
if (!isRunning()) {
return;
}
- this.shutdown = true;
+ this.shutdownStarted = true;
logger.info(LocalizedMessage.create(
LocalizedStrings.AcceptorImpl_CACHE_SERVER_ON_PORT_0_IS_SHUTTING_DOWN, this.localPort));
if (this.thread != null) {
@@ -1572,81 +1542,87 @@ public class AcceptorImpl extends Acceptor implements Runnable {
try {
this.serverSock.close();
} catch (IOException ignore) {
- // Well, we tried. Continue shutting down.
}
+
crHelper.setShutdown(true); // set this before shutting down the pool
- if (isSelector()) {
- this.hsTimer.cancel();
- if (this.tmpSel != null) {
- try {
- this.tmpSel.close();
- } catch (IOException ignore) {
- }
- }
- try {
- wakeupSelector();
- this.selector.close();
- } catch (IOException ignore) {
- }
- if (this.selectorThread != null) {
- this.selectorThread.interrupt();
- }
- this.commBufferQueue.clear();
- }
+ shutdownSelectorIfIsSelector();
ClientHealthMonitor.shutdownInstance();
shutdownSCs();
this.clientNotifier.shutdown(this.acceptorId);
- this.pool.shutdown();
-
- try {
- if (!this.pool.awaitTermination(PoolImpl.SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS)) {
- logger.warn(LocalizedMessage.create(
- LocalizedStrings.PoolImpl_TIMEOUT_WAITING_FOR_BACKGROUND_TASKS_TO_COMPLETE));
- this.pool.shutdownNow();
- }
- } catch (InterruptedException ignore) {
- Thread.currentThread().interrupt();
- this.pool.shutdownNow();
- }
- this.hsPool.shutdownNow();
+ shutdownPools();
this.stats.close();
- GemFireCacheImpl myCache = (GemFireCacheImpl) cache;
- if (!myCache.forcedDisconnect()) {
- Set<PartitionedRegion> prs = myCache.getPartitionedRegions();
- for (PartitionedRegion pr : prs) {
- Map<Integer, BucketAdvisor.BucketProfile> profiles =
- new HashMap<Integer, BucketAdvisor.BucketProfile>();
- // 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 receipients = new HashSet();
- receipients = pr.getRegionAdvisor().adviseAllPRNodes();
- // send it to all in one messgae
- ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(receipients,
- pr.getDistributionManager(), pr.getPRId(), profiles, true);
- if (reply != null) {
- reply.waitForRepliesUninterruptibly();
- }
-
- if (logger.isDebugEnabled()) {
- logger.debug("sending messages to all peers for removing this server..");
- }
- }
- }
- acceptorImplObserver.normalCloseTermination(this);
+ notifyCacheMembersOfClose();
} // synchronized
} catch (RuntimeException e) {/* ignore and log */
logger.warn(LocalizedMessage.create(LocalizedStrings.AcceptorImpl_UNEXPECTED), e);
- } finally {
- acceptorImplObserver.afterClose(this);
}
}
+ private void notifyCacheMembersOfClose() {
+ GemFireCacheImpl myCache = (GemFireCacheImpl) cache;
+ if (!myCache.forcedDisconnect()) {
+ for (PartitionedRegion pr : myCache.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();
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("sending messages to all peers for removing this server..");
+ }
+ }
+ }
+ }
+
+ private void shutdownSelectorIfIsSelector() {
+ if (isSelector()) {
+ this.hsTimer.cancel();
+ if (this.tmpSel != null) {
+ try {
+ this.tmpSel.close();
+ } catch (IOException ignore) {
+ }
+ }
+ try {
+ wakeupSelector();
+ this.selector.close();
+ } catch (IOException ignore) {
+ }
+ if (this.selectorThread != null) {
+ this.selectorThread.interrupt();
+ }
+ this.commBufferQueue.clear();
+ }
+ }
+
+ private void shutdownPools() {
+ this.pool.shutdown();
+ try {
+ if (!this.pool.awaitTermination(PoolImpl.SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS)) {
+ logger.warn(LocalizedMessage
+ .create(LocalizedStrings.PoolImpl_TIMEOUT_WAITING_FOR_BACKGROUND_TASKS_TO_COMPLETE));
+ this.pool.shutdownNow();
+ }
+ } catch (InterruptedException ignore) {
+ Thread.currentThread().interrupt();
+ this.pool.shutdownNow();
+ }
+ this.hsPool.shutdownNow();
+ }
+
private void shutdownSCs() {
// added to fix part 2 of bug 37351.
synchronized (this.allSCsLock) {
@@ -1657,36 +1633,12 @@ public class AcceptorImpl extends Acceptor implements Runnable {
}
}
+ public boolean isShutdownProperly() {
+ return !isRunning() && (selectorThread == null || !selectorThread.isAlive())
+ && (pool == null || pool.isShutdown()) && (hsPool == null || hsPool.isShutdown())
+ && (selector == null || !selector.isOpen());
+ }
- // protected InetAddress getBindAddress() {
- // return this.bindAddress;
- // }
-
- // /**
- // * Calculates the bind address based on gemfire.properties.
- // * Returns null if no bind address is configured.
- // * @since GemFire 5.7
- // */
- // public static InetAddress calcBindAddress(Cache cache) throws IOException {
- // InternalDistributedSystem system = (InternalDistributedSystem)cache
- // .getDistributedSystem();
- // DistributionConfig config = system.getConfig();
- // InetAddress address = null;
-
- // // Get the server-bind-address. If it is not null, use it.
- // // If it is null, get the bind-address. If it is not null, use it.
- // // Otherwise set default.
- // String serverBindAddress = config.getServerBindAddress();
- // if (serverBindAddress != null && serverBindAddress.length() > 0) {
- // address = InetAddress.getByName(serverBindAddress);
- // } else {
- // String bindAddress = config.getBindAddress();
- // if (bindAddress != null && bindAddress.length() > 0) {
- // address = InetAddress.getByName(bindAddress);
- // }
- // }
- // return address;
- // }
/**
* @param bindName the ip address or host name that this acceptor should bind to. If null or ""
* then calculate it.
http://git-wip-us.apache.org/repos/asf/geode/blob/d73ec978/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
deleted file mode 100644
index 3d02878..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.tier.sockets.command;
-
-import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
-
-/**
- * AcceptorImplObserver is an observer/visitor for AcceptorImpl that is used for testing.
- */
-public interface AcceptorImplObserver {
- void beforeClose(AcceptorImpl acceptorImpl);
-
- void normalCloseTermination(AcceptorImpl acceptorImpl);
-
- void afterClose(AcceptorImpl acceptorImpl);
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/d73ec978/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
index ca8592f..810aabc 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
@@ -15,9 +15,10 @@
package org.apache.geode.internal.cache.tier.sockets;
-import org.apache.geode.cache.Cache;
+import com.jayway.awaitility.Awaitility;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
@@ -27,22 +28,21 @@ import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheWriterAdapter;
import org.apache.geode.internal.AvailablePortHelper;
+import org.apache.geode.internal.cache.CacheServerImpl;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.tier.sockets.command.AcceptorImplObserver;
import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.DistributedTest;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import java.lang.management.ManagementFactory;
-import java.lang.management.ThreadInfo;
-import java.lang.management.ThreadMXBean;
-import java.util.Arrays;
+import java.util.List;
import java.util.Properties;
-import java.util.concurrent.atomic.AtomicBoolean;
+import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.junit.Assert.*;
@@ -52,44 +52,79 @@ import static org.junit.Assert.*;
*/
@Category(DistributedTest.class)
public class AcceptorImplDUnitTest extends JUnit4DistributedTestCase {
- private static Cache cache;
public AcceptorImplDUnitTest() {
super();
}
- @Override
- public void postTearDown() throws Exception {
- if (cache != null) {
- cache.close();
- cache = null;
+ // SleepyCacheWriter will block indefinitely.
+ // Anyone who has a handle on the SleepyCacheWriter can interrupt it by calling wakeUp.
+ class SleepyCacheWriter<K, V> extends CacheWriterAdapter<K, V> {
+ private boolean setOnStart;
+ private boolean setOnInterrupt;
+ private boolean stopWaiting;
+ // locks the above three booleans.
+ private final Object lock = new Object();
+
+ public void notifyStart() {
+ synchronized (lock) {
+ setOnStart = true;
+ }
+ }
+
+ public boolean isStarted() {
+ synchronized (lock) {
+ return setOnStart;
+ }
+ }
+
+ public void notifyInterrupt() {
+ synchronized (lock) {
+ setOnInterrupt = true;
+ }
+ }
+
+ public boolean isInterrupted() {
+ synchronized (lock) {
+ return setOnInterrupt;
+ }
+ }
+
+ public void stopWaiting() {
+ synchronized (lock) {
+ this.stopWaiting = true;
+ lock.notify();
+ }
+ }
+
+ public boolean isReadyToQuit() {
+ synchronized (lock) {
+ return stopWaiting;
+ }
}
- super.postTearDown();
- }
- public static class SleepyCacheWriter<K, V> extends CacheWriterAdapter<K, V> {
+ SleepyCacheWriter() {}
+
@Override
public void beforeCreate(EntryEvent<K, V> event) {
- while (true) {
- System.out.println("Sleeping a long time.");
+ System.out.println("Sleeping a long time.");
+ notifyStart();
+ while (!isReadyToQuit()) {
try {
- Thread.sleep(100000000);
- } catch (InterruptedException ignore) {
+ synchronized (lock) {
+ lock.wait();
+ }
+ } catch (InterruptedException ex) {
+ notifyInterrupt();
}
}
+ if (isInterrupted()) {
+ Thread.currentThread().interrupt();
+ }
}
}
/**
- * Dump threads to standard out. For debugging.
- */
- private void dumpThreads() {
- ThreadMXBean bean = ManagementFactory.getThreadMXBean();
- ThreadInfo[] infos = bean.dumpAllThreads(true, true);
- System.out.println("infos = " + Arrays.toString(infos));
- }
-
- /**
* GEODE-2324. There was a bug where, due to an uncaught exception, `AcceptorImpl.close()` was
* short-circuiting and failing to clean up properly.
*
@@ -100,40 +135,19 @@ public class AcceptorImplDUnitTest extends JUnit4DistributedTestCase {
* since the fields are private) and implementation-dependent.
*/
@Test
- public void testShutdownCatchesException() throws Exception {
+ public void testAcceptorImplCloseCleansUpWithHangingConnection() throws Exception {
final String hostname = Host.getHost(0).getHostName();
final VM clientVM = Host.getHost(0).getVM(0);
- // AtomicBooleans can be set from wherever they are, including an anonymous class or other
- // thread.
- AtomicBoolean terminatedNormally = new AtomicBoolean(false);
- AtomicBoolean passedPostConditions = new AtomicBoolean(false);
-
Properties props = new Properties();
props.setProperty(MCAST_PORT, "0");
- AcceptorImpl.setObserver_TESTONLY(new AcceptorImplObserver() {
- @Override
- public void beforeClose(AcceptorImpl acceptorImpl) {
- Thread.currentThread().interrupt();
- }
-
- @Override
- public void normalCloseTermination(AcceptorImpl acceptorImpl) {
- terminatedNormally.set(true);
- }
-
- @Override
- public void afterClose(AcceptorImpl acceptorImpl) {
- passedPostConditions.set(!acceptorImpl.isRunning());
- }
- });
-
try (InternalCache cache = (InternalCache) new CacheFactory(props).create()) {
RegionFactory<Object, Object> regionFactory =
cache.createRegionFactory(RegionShortcut.PARTITION);
- regionFactory.setCacheWriter(new SleepyCacheWriter<>());
+ SleepyCacheWriter<Object, Object> sleepyCacheWriter = new SleepyCacheWriter<>();
+ regionFactory.setCacheWriter(sleepyCacheWriter);
final CacheServer server = cache.addCacheServer();
final int port = AvailablePortHelper.getRandomAvailableTCPPort();
@@ -142,32 +156,93 @@ public class AcceptorImplDUnitTest extends JUnit4DistributedTestCase {
regionFactory.create("region1");
+ assertTrue(cache.isServer());
+ assertFalse(cache.isClosed());
+
+ Awaitility.await("Acceptor is up and running").atMost(10, SECONDS)
+ .until(() -> getAcceptorImplFromCache(cache) != null);
+ AcceptorImpl acceptorImpl = getAcceptorImplFromCache(cache);
+
+
clientVM.invokeAsync(() -> {
+ // System.setProperty("gemfire.PoolImpl.TRY_SERVERS_ONCE", "true");
ClientCacheFactory clientCacheFactory = new ClientCacheFactory();
clientCacheFactory.addPoolServer(hostname, port);
+ clientCacheFactory.setPoolReadTimeout(5000);
+ clientCacheFactory.setPoolRetryAttempts(1);
+ clientCacheFactory.setPoolMaxConnections(1);
+ clientCacheFactory.setPoolFreeConnectionTimeout(1000);
ClientCache clientCache = clientCacheFactory.create();
Region<Object, Object> clientRegion1 =
clientCache.createClientRegionFactory(ClientRegionShortcut.PROXY).create("region1");
clientRegion1.put("foo", "bar");
});
+ Awaitility.await("Cache writer starts").atMost(10, SECONDS)
+ .until(sleepyCacheWriter::isStarted);
+
cache.close();
- dumpThreads();
- assertTrue(terminatedNormally.get());
- assertTrue(passedPostConditions.get());
+ Awaitility.await("Cache writer interrupted").atMost(10, SECONDS)
+ .until(sleepyCacheWriter::isInterrupted);
- // cleanup.
- AcceptorImpl.setObserver_TESTONLY(new AcceptorImplObserver() {
- @Override
- public void beforeClose(AcceptorImpl acceptorImpl) {}
+ sleepyCacheWriter.stopWaiting();
- @Override
- public void normalCloseTermination(AcceptorImpl acceptorImpl) {}
+ Awaitility.await("Acceptor shuts down properly").atMost(10, SECONDS)
+ .until(() -> acceptorImpl.isShutdownProperly());
- @Override
- public void afterClose(AcceptorImpl acceptorImpl) {}
- });
+ ThreadUtils.dumpMyThreads(); // for debugging.
+
+ regionFactory.setCacheWriter(null);
+ }
+ }
+
+
+ @Test
+ public void testAcceptorImplCloseCleansUp() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(MCAST_PORT, "0");
+
+ try (InternalCache cache = (InternalCache) new CacheFactory(props).create()) {
+ RegionFactory<Object, Object> regionFactory =
+ cache.createRegionFactory(RegionShortcut.PARTITION);
+
+ final CacheServer server = cache.addCacheServer();
+ final int port = AvailablePortHelper.getRandomAvailableTCPPort();
+ server.setPort(port);
+ server.start();
+
+ regionFactory.create("region1");
+
+ assertTrue(cache.isServer());
+ assertFalse(cache.isClosed());
+ Awaitility.await("Acceptor is up and running").atMost(10, SECONDS)
+ .until(() -> getAcceptorImplFromCache(cache) != null);
+
+ AcceptorImpl acceptorImpl = getAcceptorImplFromCache(cache);
+
+ cache.close();
+ Awaitility.await("Acceptor shuts down properly").atMost(10, SECONDS)
+ .until(acceptorImpl::isShutdownProperly);
+
+ assertTrue(cache.isClosed());
+ assertFalse(acceptorImpl.isRunning());
+ }
+ }
+
+ /**
+ *
+ * @param cache
+ * @return the cache's Acceptor, if there is exactly one CacheServer. Otherwise null.
+ */
+ public AcceptorImpl getAcceptorImplFromCache(GemFireCache cache) {
+ GemFireCacheImpl gemFireCache = (GemFireCacheImpl) cache;
+ List<CacheServer> cacheServers = gemFireCache.getCacheServers();
+ if (cacheServers.size() != 1) {
+ return null;
}
+
+ CacheServerImpl cacheServerImpl = (CacheServerImpl) cacheServers.get(0);
+ return cacheServerImpl.getAcceptor();
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/d73ec978/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
index 58c2157..7aa11b7 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplJUnitTest.java
@@ -15,13 +15,9 @@
package org.apache.geode.internal.cache.tier.sockets;
import org.apache.geode.cache.AttributesFactory;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
-import org.apache.geode.cache.RegionFactory;
-import org.apache.geode.cache.client.ClientCache;
-import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ServerRefusedConnectionException;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedSystem;
@@ -46,16 +42,30 @@ import java.net.BindException;
import java.net.Socket;
import java.util.Collections;
import java.util.Properties;
-import java.util.Set;
-import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
-import static org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID.system;
import static org.junit.Assert.*;
@Category({IntegrationTest.class, ClientServerTest.class})
public class AcceptorImplJUnitTest {
+ DistributedSystem system;
+ InternalCache cache;
+
+ @Before
+ public void setUp() throws Exception {
+ Properties p = new Properties();
+ p.setProperty(MCAST_PORT, "0");
+ this.system = DistributedSystem.connect(p);
+ this.cache = (InternalCache) CacheFactory.create(system);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ this.cache.close();
+ this.system.disconnect();
+ }
+
/*
* Test method for 'org.apache.geode.internal.cache.tier.sockets.AcceptorImpl(int, int, boolean,
* int, Cache)'
@@ -64,16 +74,15 @@ public class AcceptorImplJUnitTest {
@Test
public void testConstructor() throws CacheException, IOException {
AcceptorImpl a1 = null, a2 = null, a3 = null;
- Properties props = new Properties();
- props.setProperty(MCAST_PORT, "0");
- try (InternalCache cache = (InternalCache) new CacheFactory(props).create()) {
+ try {
final int[] freeTCPPorts = AvailablePortHelper.getRandomAvailableTCPPorts(2);
int port1 = freeTCPPorts[0];
int port2 = freeTCPPorts[1];
+
try {
new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, cache,
+ 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);
@@ -83,7 +92,7 @@ public class AcceptorImplJUnitTest {
try {
new AcceptorImpl(port2, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, cache, 0,
+ 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);
@@ -93,12 +102,12 @@ public class AcceptorImplJUnitTest {
try {
a1 = new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, cache,
+ 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);
a2 = new AcceptorImpl(port1, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, cache,
+ 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);
@@ -107,12 +116,13 @@ public class AcceptorImplJUnitTest {
}
a3 = new AcceptorImpl(port2, null, false, CacheServer.DEFAULT_SOCKET_BUFFER_SIZE,
- CacheServer.DEFAULT_MAXIMUM_TIME_BETWEEN_PINGS, cache,
+ 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);
assertEquals(port2, a3.getPort());
- InternalDistributedSystem isystem = (InternalDistributedSystem) cache.getDistributedSystem();
+ InternalDistributedSystem isystem =
+ (InternalDistributedSystem) this.cache.getDistributedSystem();
DistributionConfig config = isystem.getConfig();
String bindAddress = config.getBindAddress();
if (bindAddress == null || bindAddress.length() <= 0) {
[05/12] geode git commit: [GEODE-2324] Make AcceptorImplObserver an
interface
Posted by kl...@apache.org.
[GEODE-2324] Make AcceptorImplObserver an interface
and fix threading issues.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/6bed2828
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/6bed2828
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/6bed2828
Branch: refs/heads/feature/GEODE-2456
Commit: 6bed28285d8ff150c1759a2b04cb9959730aae51
Parents: 1d561dc
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Tue Jan 24 15:28:25 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../cache/tier/sockets/AcceptorImpl.java | 41 +++++++++++++++-----
.../sockets/command/AcceptorImplObserver.java | 26 ++-----------
.../tier/sockets/AcceptorImplDUnitTest.java | 13 ++++++-
3 files changed, 46 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/6bed2828/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 d72c72b..1bca1fd 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
@@ -271,6 +271,33 @@ public class AcceptorImpl extends Acceptor implements Runnable {
private SecurityService securityService = IntegratedSecurityService.getSecurityService();
+ // Assumed non-null. Do not set this to null.
+ private static AcceptorImplObserver acceptorImplObserver_do_not_access_directly =
+ new AcceptorImplObserver() {
+ @Override
+ public void beforeClose(AcceptorImpl acceptorImpl) {}
+
+ @Override
+ public void normalCloseTermination(AcceptorImpl acceptorImpl) {}
+
+ @Override
+ public void afterClose(AcceptorImpl acceptorImpl) {}
+ };
+
+ private static AcceptorImplObserver getAcceptorImplObserver() {
+ synchronized (AcceptorImpl.class) {
+ return acceptorImplObserver_do_not_access_directly;
+ }
+ }
+
+ public static void setObserver_TESTONLY(AcceptorImplObserver observer) {
+ synchronized (AcceptorImpl.class) {
+ if (observer != null) {
+ acceptorImplObserver_do_not_access_directly = observer;
+ }
+ }
+ }
+
/**
* Initializes this acceptor thread to listen for connections on the given port.
*
@@ -1529,12 +1556,10 @@ public class AcceptorImpl extends Acceptor implements Runnable {
@Override
public void close() {
- AcceptorImplObserver acceptorImplObserver = AcceptorImplObserver.getInstance();
+ AcceptorImplObserver acceptorImplObserver = getAcceptorImplObserver();
try {
synchronized (syncLock) {
- if (acceptorImplObserver != null) {
- acceptorImplObserver.beforeClose(this);
- }
+ acceptorImplObserver.beforeClose(this);
if (!isRunning()) {
return;
}
@@ -1613,16 +1638,12 @@ public class AcceptorImpl extends Acceptor implements Runnable {
}
}
}
- if (acceptorImplObserver != null) {
- acceptorImplObserver.normalCloseTermination(this);
- }
+ acceptorImplObserver.normalCloseTermination(this);
} // synchronized
} catch (RuntimeException e) {/* ignore and log */
logger.warn(LocalizedMessage.create(LocalizedStrings.AcceptorImpl_UNEXPECTED), e);
} finally {
- if (acceptorImplObserver != null) {
- acceptorImplObserver.afterClose(this);
- }
+ acceptorImplObserver.afterClose(this);
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/6bed2828/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
index f5ee982..3d02878 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/AcceptorImplObserver.java
@@ -20,28 +20,10 @@ import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
/**
* AcceptorImplObserver is an observer/visitor for AcceptorImpl that is used for testing.
*/
-public abstract class AcceptorImplObserver {
- private static AcceptorImplObserver instance;
+public interface AcceptorImplObserver {
+ void beforeClose(AcceptorImpl acceptorImpl);
- /**
- * Set the instance of the observer. Setting to null will clear the observer.
- *
- * @param instance
- * @return the old observer, or null if there was no old observer.
- */
- public static final AcceptorImplObserver setInstance(AcceptorImplObserver instance) {
- AcceptorImplObserver oldInstance = AcceptorImplObserver.instance;
- AcceptorImplObserver.instance = instance;
- return oldInstance;
- }
+ void normalCloseTermination(AcceptorImpl acceptorImpl);
- public static final AcceptorImplObserver getInstance() {
- return instance;
- }
-
- public void beforeClose(AcceptorImpl acceptorImpl) {}
-
- public void normalCloseTermination(AcceptorImpl acceptorImpl) {}
-
- public void afterClose(AcceptorImpl acceptorImpl) {}
+ void afterClose(AcceptorImpl acceptorImpl);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/6bed2828/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
index 8b4c672..ca8592f 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImplDUnitTest.java
@@ -112,7 +112,7 @@ public class AcceptorImplDUnitTest extends JUnit4DistributedTestCase {
Properties props = new Properties();
props.setProperty(MCAST_PORT, "0");
- AcceptorImplObserver.setInstance(new AcceptorImplObserver() {
+ AcceptorImpl.setObserver_TESTONLY(new AcceptorImplObserver() {
@Override
public void beforeClose(AcceptorImpl acceptorImpl) {
Thread.currentThread().interrupt();
@@ -158,7 +158,16 @@ public class AcceptorImplDUnitTest extends JUnit4DistributedTestCase {
assertTrue(passedPostConditions.get());
// cleanup.
- AcceptorImplObserver.setInstance(null);
+ AcceptorImpl.setObserver_TESTONLY(new AcceptorImplObserver() {
+ @Override
+ public void beforeClose(AcceptorImpl acceptorImpl) {}
+
+ @Override
+ public void normalCloseTermination(AcceptorImpl acceptorImpl) {}
+
+ @Override
+ public void afterClose(AcceptorImpl acceptorImpl) {}
+ });
}
}
}
[03/12] geode git commit: [GEODE-2324] fixes to AcceptorImpl.close()
Posted by kl...@apache.org.
[GEODE-2324] fixes to AcceptorImpl.close()
If the thread is interrupted during closing, just continue to shut down
what we can.
* Catch InterruptedException so cleanup continues.
* Remove top-level exception handler to avoid masking exceptions that
* could short-circuit shutdown.
* Fix a synchronization bug that could cause AcceptorImpl to try to shut
down twice.
* Fix what looks like a bug where if closing the socket throws an
IOException, we fail to shut anything else down, though we still have
ourselves marked as shut down.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/75b02565
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/75b02565
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/75b02565
Branch: refs/heads/feature/GEODE-2456
Commit: 75b02565b272a4b1dc5763f768bfed26fbed3d18
Parents: 30341ec
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Wed Jan 18 15:00:47 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../cache/tier/sockets/AcceptorImpl.java | 133 +++++++++----------
1 file changed, 65 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/75b02565/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 060683d..ea6b369 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
@@ -1527,89 +1527,86 @@ public class AcceptorImpl extends Acceptor implements Runnable {
}
@Override
- @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "REC_CATCH_EXCEPTION",
- justification = "Allow this thread to die")
public void close() {
- if (!isRunning()) {
- return;
- }
-
- try {
- synchronized (syncLock) {
- this.shutdown = true;
- logger.info(LocalizedMessage.create(
- LocalizedStrings.AcceptorImpl_CACHE_SERVER_ON_PORT_0_IS_SHUTTING_DOWN, this.localPort));
- if (this.thread != null) {
- this.thread.interrupt();
- }
+ synchronized (syncLock) {
+ if (!isRunning()) {
+ return;
+ }
+ this.shutdown = true;
+ logger.info(LocalizedMessage.create(
+ LocalizedStrings.AcceptorImpl_CACHE_SERVER_ON_PORT_0_IS_SHUTTING_DOWN, this.localPort));
+ if (this.thread != null) {
+ this.thread.interrupt();
+ }
+ try {
this.serverSock.close();
- crHelper.setShutdown(true); // set this before shutting down the pool
- if (isSelector()) {
- this.hsTimer.cancel();
- if (this.tmpSel != null) {
- try {
- this.tmpSel.close();
- } catch (IOException ignore) {
- }
- }
+ } catch (IOException ignore) {
+ // Well, we tried. Continue shutting down.
+ }
+ crHelper.setShutdown(true); // set this before shutting down the pool
+ if (isSelector()) {
+ this.hsTimer.cancel();
+ if (this.tmpSel != null) {
try {
- wakeupSelector();
- this.selector.close();
+ this.tmpSel.close();
} catch (IOException ignore) {
}
- if (this.selectorThread != null) {
- this.selectorThread.interrupt();
- }
- this.commBufferQueue.clear();
}
- ClientHealthMonitor.shutdownInstance();
- shutdownSCs();
- this.clientNotifier.shutdown(this.acceptorId);
- this.pool.shutdown();
+ try {
+ wakeupSelector();
+ this.selector.close();
+ } catch (IOException ignore) {
+ }
+ if (this.selectorThread != null) {
+ this.selectorThread.interrupt();
+ }
+ this.commBufferQueue.clear();
+ }
+ ClientHealthMonitor.shutdownInstance();
+ shutdownSCs();
+ this.clientNotifier.shutdown(this.acceptorId);
+ this.pool.shutdown();
+
+ try {
if (!this.pool.awaitTermination(PoolImpl.SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS)) {
logger.warn(LocalizedMessage
.create(LocalizedStrings.PoolImpl_TIMEOUT_WAITING_FOR_BACKGROUND_TASKS_TO_COMPLETE));
this.pool.shutdownNow();
}
- this.hsPool.shutdownNow();
- this.stats.close();
- GemFireCacheImpl myCache = (GemFireCacheImpl) cache;
- if (!myCache.forcedDisconnect()) {
- Set<PartitionedRegion> prs = myCache.getPartitionedRegions();
- for (PartitionedRegion pr : prs) {
- Map<Integer, BucketAdvisor.BucketProfile> profiles =
- new HashMap<Integer, BucketAdvisor.BucketProfile>();
- // 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 receipients = new HashSet();
- receipients = pr.getRegionAdvisor().adviseAllPRNodes();
- // send it to all in one messgae
- ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(receipients,
- pr.getDistributionManager(), pr.getPRId(), profiles, true);
- if (reply != null) {
- reply.waitForRepliesUninterruptibly();
- }
+ } catch (InterruptedException ignore) {
+ this.pool.shutdownNow();
+ }
+ this.hsPool.shutdownNow();
+ this.stats.close();
+ GemFireCacheImpl myCache = (GemFireCacheImpl) cache;
+ if (!myCache.forcedDisconnect()) {
+ Set<PartitionedRegion> prs = myCache.getPartitionedRegions();
+ for (PartitionedRegion pr : prs) {
+ Map<Integer, BucketAdvisor.BucketProfile> profiles =
+ new HashMap<Integer, BucketAdvisor.BucketProfile>();
+ // 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 receipients = new HashSet();
+ receipients = pr.getRegionAdvisor().adviseAllPRNodes();
+ // send it to all in one messgae
+ ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(receipients,
+ pr.getDistributionManager(), pr.getPRId(), profiles, true);
+ if (reply != null) {
+ reply.waitForRepliesUninterruptibly();
+ }
- if (logger.isDebugEnabled()) {
- logger.debug("sending messages to all peers for removing this server..");
- }
+ if (logger.isDebugEnabled()) {
+ logger.debug("sending messages to all peers for removing this server..");
}
}
-
- } // synchronized
- } catch (InterruptedException e) {
- if (!this.shutdown) { // GEODE-1103: expect an interrupt during shutdown
- logger.warn(LocalizedMessage.create(LocalizedStrings.AcceptorImpl_UNEXPECTED), e);
}
- } catch (Exception e) {/* ignore */
- logger.warn(LocalizedMessage.create(LocalizedStrings.AcceptorImpl_UNEXPECTED), e);
- }
+ } // synchronized
}
private void shutdownSCs() {
[06/12] geode git commit: [GEODE-2324] Re-interrupt after catching
InterruptedException.
Posted by kl...@apache.org.
[GEODE-2324] Re-interrupt after catching InterruptedException.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/cae580f4
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/cae580f4
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/cae580f4
Branch: refs/heads/feature/GEODE-2456
Commit: cae580f4b173fad1f38a618989561701aba98e74
Parents: 7b10be4
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Fri Jan 20 15:21:30 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/cae580f4/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 05f8835..37438b6 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
@@ -1575,6 +1575,7 @@ public class AcceptorImpl extends Acceptor implements Runnable {
this.pool.shutdownNow();
}
} catch (InterruptedException ignore) {
+ Thread.currentThread().interrupt();
this.pool.shutdownNow();
}
this.hsPool.shutdownNow();
[04/12] geode git commit: [GEODE-2324] Reinstate Exception catching
for RuntimeException
Posted by kl...@apache.org.
[GEODE-2324] Reinstate Exception catching for RuntimeException
so we gain the benefits of logging unchecked exceptions without losing
the type system benefits of having to catch exceptions.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/7b10be4c
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/7b10be4c
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/7b10be4c
Branch: refs/heads/feature/GEODE-2456
Commit: 7b10be4c6295d53b66d26e0461a04b6feecc0ee7
Parents: 75b0256
Author: Galen O'Sullivan <go...@pivotal.io>
Authored: Fri Jan 20 11:37:14 2017 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Thu Feb 9 14:28:48 2017 -0800
----------------------------------------------------------------------
.../cache/tier/sockets/AcceptorImpl.java | 140 ++++++++++---------
1 file changed, 72 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/7b10be4c/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
----------------------------------------------------------------------
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 ea6b369..05f8835 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
@@ -1528,85 +1528,89 @@ public class AcceptorImpl extends Acceptor implements Runnable {
@Override
public void close() {
- synchronized (syncLock) {
- if (!isRunning()) {
- return;
- }
- this.shutdown = true;
- logger.info(LocalizedMessage.create(
- LocalizedStrings.AcceptorImpl_CACHE_SERVER_ON_PORT_0_IS_SHUTTING_DOWN, this.localPort));
- if (this.thread != null) {
- this.thread.interrupt();
- }
- try {
- this.serverSock.close();
- } catch (IOException ignore) {
- // Well, we tried. Continue shutting down.
- }
- crHelper.setShutdown(true); // set this before shutting down the pool
- if (isSelector()) {
- this.hsTimer.cancel();
- if (this.tmpSel != null) {
- try {
- this.tmpSel.close();
- } catch (IOException ignore) {
- }
+ try {
+ synchronized (syncLock) {
+ if (!isRunning()) {
+ return;
+ }
+ this.shutdown = true;
+ logger.info(LocalizedMessage.create(
+ LocalizedStrings.AcceptorImpl_CACHE_SERVER_ON_PORT_0_IS_SHUTTING_DOWN, this.localPort));
+ if (this.thread != null) {
+ this.thread.interrupt();
}
try {
- wakeupSelector();
- this.selector.close();
+ this.serverSock.close();
} catch (IOException ignore) {
+ // Well, we tried. Continue shutting down.
}
- if (this.selectorThread != null) {
- this.selectorThread.interrupt();
+ crHelper.setShutdown(true); // set this before shutting down the pool
+ if (isSelector()) {
+ this.hsTimer.cancel();
+ if (this.tmpSel != null) {
+ try {
+ this.tmpSel.close();
+ } catch (IOException ignore) {
+ }
+ }
+ try {
+ wakeupSelector();
+ this.selector.close();
+ } catch (IOException ignore) {
+ }
+ if (this.selectorThread != null) {
+ this.selectorThread.interrupt();
+ }
+ this.commBufferQueue.clear();
}
- this.commBufferQueue.clear();
- }
- ClientHealthMonitor.shutdownInstance();
- shutdownSCs();
- this.clientNotifier.shutdown(this.acceptorId);
- this.pool.shutdown();
+ ClientHealthMonitor.shutdownInstance();
+ shutdownSCs();
+ this.clientNotifier.shutdown(this.acceptorId);
+ this.pool.shutdown();
- try {
- if (!this.pool.awaitTermination(PoolImpl.SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS)) {
- logger.warn(LocalizedMessage
- .create(LocalizedStrings.PoolImpl_TIMEOUT_WAITING_FOR_BACKGROUND_TASKS_TO_COMPLETE));
+ try {
+ if (!this.pool.awaitTermination(PoolImpl.SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS)) {
+ logger.warn(LocalizedMessage.create(
+ LocalizedStrings.PoolImpl_TIMEOUT_WAITING_FOR_BACKGROUND_TASKS_TO_COMPLETE));
+ this.pool.shutdownNow();
+ }
+ } catch (InterruptedException ignore) {
this.pool.shutdownNow();
}
- } catch (InterruptedException ignore) {
- this.pool.shutdownNow();
- }
- this.hsPool.shutdownNow();
- this.stats.close();
- GemFireCacheImpl myCache = (GemFireCacheImpl) cache;
- if (!myCache.forcedDisconnect()) {
- Set<PartitionedRegion> prs = myCache.getPartitionedRegions();
- for (PartitionedRegion pr : prs) {
- Map<Integer, BucketAdvisor.BucketProfile> profiles =
- new HashMap<Integer, BucketAdvisor.BucketProfile>();
- // 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 receipients = new HashSet();
- receipients = pr.getRegionAdvisor().adviseAllPRNodes();
- // send it to all in one messgae
- ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(receipients,
- pr.getDistributionManager(), pr.getPRId(), profiles, true);
- if (reply != null) {
- reply.waitForRepliesUninterruptibly();
- }
+ this.hsPool.shutdownNow();
+ this.stats.close();
+ GemFireCacheImpl myCache = (GemFireCacheImpl) cache;
+ if (!myCache.forcedDisconnect()) {
+ Set<PartitionedRegion> prs = myCache.getPartitionedRegions();
+ for (PartitionedRegion pr : prs) {
+ Map<Integer, BucketAdvisor.BucketProfile> profiles =
+ new HashMap<Integer, BucketAdvisor.BucketProfile>();
+ // 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 receipients = new HashSet();
+ receipients = pr.getRegionAdvisor().adviseAllPRNodes();
+ // send it to all in one messgae
+ ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(receipients,
+ pr.getDistributionManager(), pr.getPRId(), profiles, true);
+ if (reply != null) {
+ reply.waitForRepliesUninterruptibly();
+ }
- if (logger.isDebugEnabled()) {
- logger.debug("sending messages to all peers for removing this server..");
+ if (logger.isDebugEnabled()) {
+ logger.debug("sending messages to all peers for removing this server..");
+ }
}
}
- }
- } // synchronized
+ } // synchronized
+ } catch (RuntimeException e) {/* ignore and log */
+ logger.warn(LocalizedMessage.create(LocalizedStrings.AcceptorImpl_UNEXPECTED), e);
+ }
}
private void shutdownSCs() {
[12/12] geode git commit: GEODE-2458: await expected JMX
notifications in verifyMemberNotifications
Posted by kl...@apache.org.
GEODE-2458: await expected JMX notifications in verifyMemberNotifications
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/7085c320
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/7085c320
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/7085c320
Branch: refs/heads/feature/GEODE-2456
Commit: 7085c3205a9fd1167c22b61d58767cb81097d14b
Parents: 15748ed
Author: Kirk Lund <kl...@apache.org>
Authored: Thu Feb 9 14:59:03 2017 -0800
Committer: Kirk Lund <kl...@apache.org>
Committed: Thu Feb 9 15:15:34 2017 -0800
----------------------------------------------------------------------
.../org/apache/geode/management/RegionManagementDUnitTest.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/7085c320/geode-core/src/test/java/org/apache/geode/management/RegionManagementDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/management/RegionManagementDUnitTest.java b/geode-core/src/test/java/org/apache/geode/management/RegionManagementDUnitTest.java
index 7ff5f02..aca279b 100644
--- a/geode-core/src/test/java/org/apache/geode/management/RegionManagementDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/RegionManagementDUnitTest.java
@@ -578,8 +578,7 @@ public class RegionManagementDUnitTest extends ManagementTestBase {
private void verifyMemberNotifications(final VM managerVM, final String regionName,
final int expectedMembers) {
managerVM.invoke("verifyMemberNotifications", () -> {
- assertThat(MEMBER_NOTIFICATIONS_REF.get()).isNotNull();
- assertThat(MEMBER_NOTIFICATIONS_REF.get()).hasSize(expectedMembers * 2);
+ await().until(() -> assertThat(MEMBER_NOTIFICATIONS_REF.get()).hasSize(expectedMembers * 2));
int regionCreatedCount = 0;
int regionDestroyedCount = 0;