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/12/26 22:25:07 UTC
[geode] branch feature/GEODE-5955 created (now 32399f0)
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a change to branch feature/GEODE-5955
in repository https://gitbox.apache.org/repos/asf/geode.git.
at 32399f0 GEODE-5955 CacheClientUpdater statistics are created and never closed on failed initialization
This branch includes the following new commits:
new 32399f0 GEODE-5955 CacheClientUpdater statistics are created and never closed on failed initialization
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-5955 CacheClientUpdater statistics are created
and never closed on failed initialization
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-5955
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 32399f0b162d0ee42f831d8a14f110e5742cec0b
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Wed Dec 26 14:21:32 2018 -0800
GEODE-5955 CacheClientUpdater statistics are created and never closed on failed initialization
Close statistics if construction of a CacheClientUpdater fails.
I had to introduce a StatisticsProvider in order to inject a mock
statistics object into the CacheClientUpdater. I also reworked the
"finally" block in the constructor to use the updater's close()
method.
---
.../internal/cache/tier/ClientSideHandshake.java | 4 ++
.../cache/tier/sockets/CacheClientUpdater.java | 75 +++++++++++---------
.../tier/sockets/CacheClientUpdaterJUnitTest.java | 81 ++++++++++++++++++++++
.../test/dunit/tests/BasicDistributedTest.java | 16 ++++-
4 files changed, 140 insertions(+), 36 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/ClientSideHandshake.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/ClientSideHandshake.java
index 14877b4..fe02d18 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/ClientSideHandshake.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/ClientSideHandshake.java
@@ -30,6 +30,10 @@ public interface ClientSideHandshake {
ClientProxyMembershipID getMembershipId();
+ default boolean isDurable() {
+ return getMembershipId().isDurable();
+ }
+
ServerQueueStatus handshakeWithSubscriptionFeed(Socket socket, boolean isPrimary)
throws IOException, AuthenticationRequiredException, AuthenticationFailedException,
ServerRefusedConnectionException, ClassNotFoundException;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
index dd73c1b..0167a95 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
@@ -113,7 +113,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
/**
* System of which we are a part
*/
- private final InternalDistributedSystem system;
+ private final DistributedSystem system;
/**
* The socket by which we communicate with the server
@@ -272,16 +272,30 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
EndpointManager eManager, Endpoint endpoint, int handshakeTimeout,
SocketCreator socketCreator) throws AuthenticationRequiredException,
AuthenticationFailedException, ServerRefusedConnectionException {
+ this(name, location, primary, ids, handshake, qManager, eManager, endpoint, handshakeTimeout,
+ socketCreator, new StatisticsProvider());
+ }
+
+ /**
+ * alternative constructor for unit tests. This constructor allows you to pass a
+ * mock StatisticsProvider
+ */
+ public CacheClientUpdater(String name, ServerLocation location, boolean primary,
+ DistributedSystem distributedSystem, ClientSideHandshake handshake, QueueManager qManager,
+ EndpointManager eManager, Endpoint endpoint, int handshakeTimeout,
+ SocketCreator socketCreator, StatisticsProvider statisticsProvider)
+ throws AuthenticationRequiredException,
+ AuthenticationFailedException, ServerRefusedConnectionException {
super(name);
- this.system = (InternalDistributedSystem) ids;
- this.isDurableClient = handshake.getMembershipId().isDurable();
+ this.system = distributedSystem;
+ this.isDurableClient = handshake.isDurable();
this.isPrimary = primary;
this.location = location;
this.qManager = qManager;
// this holds the connection which this threads reads
this.eManager = eManager;
this.endpoint = endpoint;
- this.stats = new CCUStats(this.system, this.location);
+ this.stats = statisticsProvider.createStatistics(distributedSystem, location);
// Create the connection...
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -398,35 +412,21 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
throw e;
} finally {
this.connected = success;
- if (mySock != null) {
- try {
- mySock.setSoTimeout(0);
- } catch (SocketException ignore) {
- // ignore: nothing we can do about this
- }
- }
-
+ this.socket = mySock;
+ this.commBuffer = cb;
+ this.out = tmpOut;
+ this.in = tmpIn;
+ this.serverId = sid;
if (this.connected) {
- this.socket = mySock;
- this.out = tmpOut;
- this.in = tmpIn;
- this.serverId = sid;
- this.commBuffer = cb;
-
- } else {
- this.socket = null;
- this.serverId = null;
- this.commBuffer = null;
- this.out = null;
- this.in = null;
-
if (mySock != null) {
try {
- mySock.close();
- } catch (IOException ioe) {
- logger.warn("Closing socket in {} failed", this, ioe);
+ mySock.setSoTimeout(0);
+ } catch (SocketException ignore) {
+ // ignore: nothing we can do about this
}
}
+ } else {
+ close();
}
}
}
@@ -471,8 +471,10 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
EntryLogger.setSource(this.serverId, "RI");
boolean addedListener = false;
try {
- this.system.addDisconnectListener(this);
- addedListener = true;
+ if (system instanceof InternalDistributedSystem) {
+ ((InternalDistributedSystem) system).addDisconnectListener(this);
+ addedListener = true;
+ }
if (!waitForCache()) {
logger.warn("{}: no cache (exiting)", this);
@@ -485,7 +487,7 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
} finally {
if (addedListener) {
- this.system.removeDisconnectListener(this);
+ ((InternalDistributedSystem) system).removeDisconnectListener(this);
}
this.close();
EntryLogger.clearSource();
@@ -547,13 +549,12 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
// ignore
}
- this.stats.close();
-
- // close the helper
if (this.cacheHelper != null) {
this.cacheHelper.close();
}
releaseCommBuffer();
+
+ this.stats.close();
}
/**
@@ -1825,6 +1826,12 @@ public class CacheClientUpdater extends LoggingThread implements ClientUpdater,
}
}
+ public static class StatisticsProvider {
+ public CCUStats createStatistics(DistributedSystem system, ServerLocation location) {
+ return new CCUStats(system, location);
+ }
+ }
+
/**
* Stats for a CacheClientUpdater. Currently the only thing measured are incoming bytes on the
* wire
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdaterJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdaterJUnitTest.java
new file mode 100644
index 0000000..d48c8a5
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdaterJUnitTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.net.SocketException;
+
+import org.junit.Test;
+
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.cache.client.internal.Endpoint;
+import org.apache.geode.cache.client.internal.EndpointManager;
+import org.apache.geode.cache.client.internal.QueueManager;
+import org.apache.geode.distributed.DistributedSystem;
+import org.apache.geode.distributed.internal.ServerLocation;
+import org.apache.geode.internal.cache.tier.ClientSideHandshake;
+import org.apache.geode.internal.net.SocketCreator;
+
+public class CacheClientUpdaterJUnitTest {
+
+ @Test
+ public void failureToConnectClosesStatistics() throws Exception {
+ // CacheClientUpdater's constructor takes a lot of parameters that we need to mock
+ ServerLocation location = new ServerLocation("localhost", 1234);
+ ClientSideHandshake handshake = mock(ClientSideHandshake.class);
+ when(handshake.isDurable()).thenReturn(Boolean.FALSE);
+ QueueManager queueManager = null;
+ mock(QueueManager.class);
+ EndpointManager endpointManager = mock(EndpointManager.class);
+ Endpoint endpoint = mock(Endpoint.class);
+
+ // shutdown checks
+ DistributedSystem distributedSystem = mock(DistributedSystem.class);
+ CancelCriterion cancelCriterion = mock(CancelCriterion.class);
+ when(distributedSystem.getCancelCriterion()).thenReturn(cancelCriterion);
+ when(cancelCriterion.isCancelInProgress()).thenReturn(Boolean.FALSE);
+
+ // engineer a failure to connect via SocketCreator
+ SocketCreator socketCreator = mock(SocketCreator.class);
+ when(socketCreator.connectForClient(any(String.class), any(Integer.class),
+ any(Integer.class), any(Integer.class))).thenThrow(new SocketException("ouch"));
+
+ // mock some stats that we can then use to ensure that they're closed when the problem occurs
+ CacheClientUpdater.StatisticsProvider statisticsProvider = mock(
+ CacheClientUpdater.StatisticsProvider.class);
+ CacheClientUpdater.CCUStats ccuStats = mock(CacheClientUpdater.CCUStats.class);
+ when(statisticsProvider
+ .createStatistics(distributedSystem, location))
+ .thenReturn(ccuStats);
+
+ // CCU's constructor fails to connect
+ CacheClientUpdater clientUpdater =
+ new CacheClientUpdater("testUpdater", location, false, distributedSystem, handshake,
+ queueManager, endpointManager, endpoint, 10000, socketCreator, statisticsProvider);
+
+ // now introspect to make sure the right actions were taken
+ // The updater should not be in a connected state
+ assertThat(clientUpdater.isConnected()).isFalse();
+ // The statistics should be closed
+ verify(ccuStats).close();
+ // The endpoint should be reported as having crashed
+ verify(endpointManager).serverCrashed(endpoint);
+ }
+}
diff --git a/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/tests/BasicDistributedTest.java b/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/tests/BasicDistributedTest.java
index 98af3d2..dafb446 100644
--- a/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/tests/BasicDistributedTest.java
+++ b/geode-dunit/src/distributedTest/java/org/apache/geode/test/dunit/tests/BasicDistributedTest.java
@@ -14,6 +14,7 @@
*/
package org.apache.geode.test.dunit.tests;
+import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.test.dunit.Invoke.invokeInEveryVM;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.hamcrest.Matchers.instanceOf;
@@ -62,6 +63,14 @@ public class BasicDistributedTest extends DistributedTestCase {
invokeInEveryVM(() -> bindings = null);
}
+ @Override
+ public Properties getDistributedSystemProperties() {
+ Properties config = super.getDistributedSystemProperties();
+ config.setProperty(LOG_LEVEL, "fine");
+ return config;
+ }
+
+
@Test
public void testPreconditions() {
invokeInEveryVM(
@@ -154,8 +163,11 @@ public class BasicDistributedTest extends DistributedTestCase {
String name = getUniqueName();
String value = "Hello";
- vm0.invokeAsync(() -> remoteBind(name, value)).join().checkException();
- vm0.invokeAsync(() -> remoteValidateBind(name, value)).join().checkException();
+ AsyncInvocation async1 = vm0.invokeAsync(() -> remoteBind(name, value)).join().checkException();
+ AsyncInvocation async2 =
+ vm0.invokeAsync(() -> remoteValidateBind(name, value)).join().checkException();
+ async1.join();
+ async2.join();
}
@Test