GEODE-2812: Add API to get list of live locators
There is a Geode cluster using a logical member group, and from the
client, the connection pool to the logical member group is connected
using the PoolFactory API at the timing when connection becomes
necessary.
At this time, even though the locator that was running at the initial
connection stops due to reasons such as regular maintenance etc., even
if the alternate locator is started before maintenance, I can not
connect to the locator in the static initial list.
1. Client side:PoolManager.createFactory().addLocator("localhost",
10334).setServerGroup("GroupA").create("pool1");
2. Geode cluster:start locator[localhost:10335].
3. Geode cluster:stop locator[localhost:10334].
4. Client side:PoolManager.createFactory().addLocator("localhost",
10334).setServerGroup("GroupB").create("pool2");
Therefore, I would like to decide the connection destination based on
the live locator list of another logical member group.
I added an API that can get the list of live locators from the Pool. Use
the API as follows:
Pool pool = PoolManager.createFactory()
.addLocator("localhost", 10334)
.setSubscriptionEnabled(true).setServerGroup("GroupA")
.create("GroupAPool");
List<InetSocketAddress> = pool.getLiveLocators();
Note:
The list of live locators gets the result of the UpdateLocatorListTask
periodically running in AutoConnectionSourceImpl.
Therefore, whether or not it is alive will cause a time lag, depending
on the task execution interval.
Also, the result of ExplicitConnectionSourceImpl without using a locator
is always empty.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/dbd7c959
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/dbd7c959
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/dbd7c959
Branch: refs/heads/develop
Commit: dbd7c959963a065d2383c042597f64e8dfb45b1f
Parents: 5f6323b
Author: masaki.yamakawa <[email protected]>
Authored: Mon Apr 24 11:20:42 2017 +0900
Committer: Bruce Schuchardt <[email protected]>
Committed: Thu May 11 09:15:27 2017 -0700
----------------------------------------------------------------------
.../org/apache/geode/cache/client/Pool.java | 12 +++++++-
.../internal/AutoConnectionSourceImpl.java | 13 ++++++++
.../cache/client/internal/ConnectionSource.java | 8 +++++
.../internal/ExplicitConnectionSourceImpl.java | 5 +++
.../geode/cache/client/internal/PoolImpl.java | 5 +++
.../geode/internal/cache/PoolFactoryImpl.java | 5 +++
.../internal/AutoConnectionSourceDUnitTest.java | 32 ++++++++++++++++++++
.../AutoConnectionSourceImplJUnitTest.java | 4 +++
.../client/internal/QueueManagerJUnitTest.java | 10 ++++++
9 files changed, 93 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/main/java/org/apache/geode/cache/client/Pool.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/Pool.java
b/geode-core/src/main/java/org/apache/geode/cache/client/Pool.java
index 16fa7a0..e37fef2 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/Pool.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/Pool.java
@@ -193,13 +193,23 @@ public interface Pool {
/**
* Returns an unmodifiable list of {@link java.net.InetSocketAddress} of the
locators this pool is
* using. Each locator is either one {@link PoolFactory#addLocator added
explicitly} when the pool
- * was created or were discovered using the explicit locators.
+ * was created.
* <p>
* If a pool has no locators then it can not discover servers or locators at
runtime.
*/
public java.util.List<InetSocketAddress> getLocators();
/**
+ * Returns an unmodifiable list of {@link java.net.InetSocketAddress} of the
locators this pool is
+ * using. The returned locator is only the currently living locator found
based on the periodic
+ * locator list request.
+ * <p>
+ * The returned locator list may be slightly old information. If the locator
does not exist, an
+ * empty list is returned.
+ */
+ public java.util.List<InetSocketAddress> getLiveLocators();
+
+ /**
* Returns an unmodifiable list of {@link java.net.InetSocketAddress} of the
servers this pool is
* using. These servers where either {@link PoolFactory#addServer added
explicitly} when the pool
* was created or were discovered using this pools {@link #getLocators
locators}.
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/main/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImpl.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImpl.java
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImpl.java
index d9fca87..a9388de 100644
---
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImpl.java
+++
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImpl.java
@@ -89,6 +89,7 @@ public class AutoConnectionSourceImpl implements
ConnectionSource {
protected final List<InetSocketAddress> initialLocators;
private final String serverGroup;
private AtomicReference<LocatorList> locators = new
AtomicReference<LocatorList>();
+ private AtomicReference<LocatorList> liveLocators = new
AtomicReference<LocatorList>();
protected InternalPool pool;
private final int connectionTimeout;
private long pingInterval;
@@ -110,6 +111,7 @@ public class AutoConnectionSourceImpl implements
ConnectionSource {
int handshakeTimeout) {
ArrayList<InetSocketAddress> tmpContacts = new
ArrayList<InetSocketAddress>(contacts);
this.locators.set(new LocatorList(tmpContacts));
+ this.liveLocators.set(new LocatorList(Collections.emptyList()));
this.initialLocators = Collections.unmodifiableList(tmpContacts);
this.connectionTimeout = handshakeTimeout;
this.serverGroup = serverGroup;
@@ -190,6 +192,14 @@ public class AutoConnectionSourceImpl implements
ConnectionSource {
return result;
}
+ @Override
+ public List<InetSocketAddress> getLiveLocators() {
+ if (PoolImpl.TEST_DURABLE_IS_NET_DOWN) {
+ return Collections.emptyList();
+ }
+ return Collections.unmodifiableList(new
ArrayList<>(liveLocators.get().getLocators()));
+ }
+
private ServerLocationResponse queryOneLocator(InetSocketAddress locator,
ServerLocationRequest request) {
@@ -253,12 +263,14 @@ public class AutoConnectionSourceImpl implements
ConnectionSource {
List<ServerLocation> locatorResponse = response.getLocators();
List<InetSocketAddress> newLocators = new
ArrayList<InetSocketAddress>(locatorResponse.size());
+ List<InetSocketAddress> activeLocators = new
ArrayList<InetSocketAddress>(locatorResponse.size());
Set<InetSocketAddress> badLocators = new
HashSet<InetSocketAddress>(initialLocators);
for (Iterator<ServerLocation> itr = locatorResponse.iterator();
itr.hasNext();) {
ServerLocation locator = itr.next();
InetSocketAddress address = new InetSocketAddress(locator.getHostName(),
locator.getPort());
newLocators.add(address);
+ activeLocators.add(address);
badLocators.remove(address);
}
@@ -287,6 +299,7 @@ public class AutoConnectionSourceImpl implements
ConnectionSource {
}
LocatorList newLocatorList = new LocatorList(newLocators);
locators.set(newLocatorList);
+ liveLocators.set(new LocatorList(activeLocators));
pool.getStats().setLocatorCount(newLocators.size());
}
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionSource.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionSource.java
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionSource.java
index eaccb6d..64bc7d8 100644
---
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionSource.java
+++
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ConnectionSource.java
@@ -14,6 +14,7 @@
*/
package org.apache.geode.cache.client.internal;
+import java.net.InetSocketAddress;
import java.util.List;
import java.util.Set;
@@ -73,4 +74,11 @@ public interface ConnectionSource {
* @return a list of all servers
*/
List<ServerLocation> getAllServers();
+
+ /**
+ * get the list of live locators
+ *
+ * @return a list of live locators
+ */
+ List<InetSocketAddress> getLiveLocators();
}
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExplicitConnectionSourceImpl.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExplicitConnectionSourceImpl.java
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExplicitConnectionSourceImpl.java
index aee69e4..1a679ce 100644
---
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExplicitConnectionSourceImpl.java
+++
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExplicitConnectionSourceImpl.java
@@ -263,4 +263,9 @@ public class ExplicitConnectionSourceImpl implements
ConnectionSource {
list.addAll(this.serverList);
return list;
}
+
+ @Override
+ public List<InetSocketAddress> getLiveLocators() {
+ return Collections.emptyList();
+ }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/main/java/org/apache/geode/cache/client/internal/PoolImpl.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PoolImpl.java
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PoolImpl.java
index 88369e1..6f63557 100644
---
a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PoolImpl.java
+++
b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PoolImpl.java
@@ -442,6 +442,11 @@ public class PoolImpl implements InternalPool {
return this.locators;
}
+ @Override
+ public List<InetSocketAddress> getLiveLocators() {
+ return this.source.getLiveLocators();
+ }
+
public List<InetSocketAddress> getServers() {
return this.servers;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/main/java/org/apache/geode/internal/cache/PoolFactoryImpl.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/PoolFactoryImpl.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/PoolFactoryImpl.java
index ef7cf03..1af748e 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/PoolFactoryImpl.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/PoolFactoryImpl.java
@@ -463,6 +463,11 @@ public class PoolFactoryImpl implements PoolFactory {
return Collections.unmodifiableList(new ArrayList(this.locators));
}
+ @Override
+ public List<InetSocketAddress> getLiveLocators() {
+ throw new UnsupportedOperationException();
+ }
+
public List/* <InetSocketAddress> */ getServers() {
if (this.locators.size() == 0 && this.servers.size() == 0) {
throw new IllegalStateException(
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceDUnitTest.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceDUnitTest.java
b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceDUnitTest.java
index 387984c..7292515 100644
---
a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceDUnitTest.java
+++
b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceDUnitTest.java
@@ -86,6 +86,7 @@ public class AutoConnectionSourceDUnitTest extends
LocatorTestBase {
vm0.invoke("StartBridgeClient",
() -> startBridgeClient(null,
NetworkUtils.getServerHostName(vm0.getHost()),
AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET)));
+ checkLocators(vm0, new InetSocketAddress[]{}, new InetSocketAddress[]{});
putInVM(vm0, "key", "value");
fail("Client cache should not have been able to start");
} catch (Exception e) {
@@ -169,22 +170,27 @@ public class AutoConnectionSourceDUnitTest extends
LocatorTestBase {
InetSocketAddress locatorToWaitFor = new InetSocketAddress(hostName,
locatorPort1);
waitForLocatorDiscovery(vm2, locatorToWaitFor);
+ InetSocketAddress[] initialLocators = new InetSocketAddress[]{new
InetSocketAddress(hostName, locatorPort0)};
+ checkLocators(vm2, initialLocators, new InetSocketAddress[]{new
InetSocketAddress(hostName, locatorPort0), new InetSocketAddress(hostName,
locatorPort1)});
vm0.invoke("Stop Locator", () -> stopLocator());
vm0.invoke("Start BridgeServer", () -> startBridgeServer(null,
locators));
putAndWaitForSuccess(vm2, REGION_NAME, "key", "value");
Assert.assertEquals("value", getInVM(vm0, "key"));
+ checkLocators(vm2, initialLocators, new InetSocketAddress[]{new
InetSocketAddress(hostName, locatorPort1)});
vm3.invoke("Start Locator", () ->
startLocator(NetworkUtils.getServerHostName(vm3.getHost()),
locatorPort3, locators));
stopBridgeMemberVM(vm0);
locatorToWaitFor = new InetSocketAddress(hostName, locatorPort3);
waitForLocatorDiscovery(vm2, locatorToWaitFor);
+ checkLocators(vm2, initialLocators, new InetSocketAddress[]{new
InetSocketAddress(hostName, locatorPort1), new InetSocketAddress(hostName,
locatorPort3)});
vm1.invoke("Stop Locator", () -> stopLocator());
vm1.invoke("Start BridgeServer", () -> startBridgeServer(null,
locators));
putAndWaitForSuccess(vm2, REGION_NAME, "key2", "value2");
Assert.assertEquals("value2", getInVM(vm1, "key2"));
+ checkLocators(vm2, initialLocators, new InetSocketAddress[]{new
InetSocketAddress(hostName, locatorPort3)});
} catch (Exception ec) {
if (ec.getCause() != null && (ec.getCause().getCause() instanceof
BindException))
return;// BindException let it pass
@@ -495,6 +501,32 @@ public class AutoConnectionSourceDUnitTest extends
LocatorTestBase {
});
}
+ protected void checkLocators(VM vm, final InetSocketAddress[]
expectedInitial, final InetSocketAddress[] expected) {
+ vm.invoke(new SerializableRunnable("Check locators") {
+ public void run() {
+ Pool pool = PoolManager.find(POOL_NAME);
+
+ List<InetSocketAddress> initialLocators = pool.getLocators();
+ Assert.assertEquals(expectedInitial.length, initialLocators.size());
+ Arrays.sort(expectedInitial,
Comparator.comparing(InetSocketAddress::getPort));
+ for (int i = 0; i < initialLocators.size(); i++) {
+ InetSocketAddress locator = initialLocators.get(i);
+ InetSocketAddress expectedOne = expectedInitial[i];
+ Assert.assertEquals(expectedOne, locator);
+ }
+
+ List<InetSocketAddress> locators = pool.getLiveLocators();
+ Assert.assertEquals(expected.length, locators.size());
+ Arrays.sort(expected,
Comparator.comparing(InetSocketAddress::getPort));
+ for (int i = 0; i < locators.size(); i++) {
+ InetSocketAddress locator = locators.get(i);
+ InetSocketAddress expectedOne = expected[i];
+ Assert.assertEquals(expectedOne, locator);
+ }
+ }
+ });
+ }
+
protected void addBridgeListener(VM vm) {
vm.invoke(new SerializableRunnable("Add membership listener") {
public void run() {
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
index 6f71dbc..1bb332f 100644
---
a/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
+++
b/geode-core/src/test/java/org/apache/geode/cache/client/internal/AutoConnectionSourceImplJUnitTest.java
@@ -368,6 +368,10 @@ public class AutoConnectionSourceImplJUnitTest {
return new ArrayList();
}
+ public List/* <InetSocketAddress> */ getLiveLocators() {
+ return new ArrayList();
+ }
+
public List/* <InetSocketAddress> */ getServers() {
return new ArrayList();
}
http://git-wip-us.apache.org/repos/asf/geode/blob/dbd7c959/geode-core/src/test/java/org/apache/geode/cache/client/internal/QueueManagerJUnitTest.java
----------------------------------------------------------------------
diff --git
a/geode-core/src/test/java/org/apache/geode/cache/client/internal/QueueManagerJUnitTest.java
b/geode-core/src/test/java/org/apache/geode/cache/client/internal/QueueManagerJUnitTest.java
index 7f9a459..081ad8c 100644
---
a/geode-core/src/test/java/org/apache/geode/cache/client/internal/QueueManagerJUnitTest.java
+++
b/geode-core/src/test/java/org/apache/geode/cache/client/internal/QueueManagerJUnitTest.java
@@ -37,6 +37,7 @@ import org.junit.experimental.categories.Category;
import java.io.InputStream;
import java.io.OutputStream;
+import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
@@ -376,6 +377,10 @@ public class QueueManagerJUnitTest {
return null;
}
+ public List getLiveLocators() {
+ return new ArrayList();
+ }
+
public int getMaxConnections() {
return 0;
}
@@ -597,6 +602,11 @@ public class QueueManagerJUnitTest {
public List<ServerLocation> getAllServers() {
return Collections.emptyList();
}
+
+ @Override
+ public List<InetSocketAddress> getLiveLocators() {
+ return Collections.emptyList();
+ }
}
private class DummyConnection implements Connection {
