patsonluk commented on code in PR #909:
URL: https://github.com/apache/solr/pull/909#discussion_r922378085
##########
solr/core/src/test/org/apache/solr/cloud/overseer/ZkStateReaderTest.java:
##########
@@ -29,171 +38,387 @@
import org.apache.solr.cloud.ZkTestServer;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.DocCollectionWatcher;
import org.apache.solr.common.cloud.DocRouter;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.util.CommonTestInjection;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
import org.apache.solr.common.util.TimeSource;
import org.apache.solr.handler.admin.ConfigSetsHandler;
import org.apache.solr.util.TimeOut;
+import org.junit.After;
+import org.junit.Before;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class ZkStateReaderTest extends SolrTestCaseJ4 {
-
+ private static final Logger log =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final long TIMEOUT = 30;
- public void testExternalCollectionWatchedNotWatched() throws Exception {
- Path zkDir = createTempDir("testExternalCollectionWatchedNotWatched");
- ZkTestServer server = new ZkTestServer(zkDir);
- SolrZkClient zkClient = null;
- ZkStateReader reader = null;
-
- try {
- server.run();
-
- zkClient = new SolrZkClient(server.getZkAddress(),
OverseerTest.DEFAULT_CONNECTION_TIMEOUT);
- ZkController.createClusterZkNodes(zkClient);
-
- reader = new ZkStateReader(zkClient);
- reader.createClusterStateWatchersAndUpdate();
-
- ZkStateWriter writer = new ZkStateWriter(reader, new Stats());
-
- zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
-
- // create new collection
- ZkWriteCommand c1 =
- new ZkWriteCommand(
- "c1",
- new DocCollection(
- "c1",
- new HashMap<>(),
- Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
- DocRouter.DEFAULT,
- 0));
- writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(c1), null);
- writer.writePendingUpdates();
- reader.forceUpdateCollection("c1");
-
-
assertTrue(reader.getClusterState().getCollectionRef("c1").isLazilyLoaded());
- reader.registerCore("c1");
-
assertFalse(reader.getClusterState().getCollectionRef("c1").isLazilyLoaded());
- reader.unregisterCore("c1");
-
assertTrue(reader.getClusterState().getCollectionRef("c1").isLazilyLoaded());
+ private static class TestFixture implements Closeable {
+ private final ZkTestServer server;
+ private final SolrZkClient zkClient;
+ private final ZkStateReader reader;
+ private final ZkStateWriter writer;
+
+ private TestFixture(
+ ZkTestServer server, SolrZkClient zkClient, ZkStateReader reader,
ZkStateWriter writer) {
+ this.server = server;
+ this.zkClient = zkClient;
+ this.reader = reader;
+ this.writer = writer;
+ }
- } finally {
+ @Override
+ public void close() throws IOException {
IOUtils.close(reader, zkClient);
- server.shutdown();
+ try {
+ server.shutdown();
+ } catch (InterruptedException e) {
+ // ok. Shutting down anyway
+ }
}
}
- public void testCollectionStateWatcherCaching() throws Exception {
- Path zkDir = createTempDir("testCollectionStateWatcherCaching");
-
- ZkTestServer server = new ZkTestServer(zkDir);
+ private TestFixture fixture = null;
- SolrZkClient zkClient = null;
- ZkStateReader reader = null;
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ fixture = setupTestFixture(getTestName());
+ }
- try {
- server.run();
-
- zkClient = new SolrZkClient(server.getZkAddress(),
OverseerTest.DEFAULT_CONNECTION_TIMEOUT);
- ZkController.createClusterZkNodes(zkClient);
-
- reader = new ZkStateReader(zkClient);
- reader.createClusterStateWatchersAndUpdate();
-
- zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
-
- ZkStateWriter writer = new ZkStateWriter(reader, new Stats());
- DocCollection state =
- new DocCollection(
- "c1",
- new HashMap<>(),
- Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
- DocRouter.DEFAULT,
- 0);
- ZkWriteCommand wc = new ZkWriteCommand("c1", state);
- writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
- writer.writePendingUpdates();
- assertTrue(zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE +
"/c1/state.json", true));
- reader.waitForState(
- "c1", 1, TimeUnit.SECONDS, (liveNodes, collectionState) ->
collectionState != null);
-
- Map<String, Object> props = new HashMap<>();
- props.put("x", "y");
- props.put(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME);
- state = new DocCollection("c1", new HashMap<>(), props,
DocRouter.DEFAULT, 0);
- wc = new ZkWriteCommand("c1", state);
- writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
- writer.writePendingUpdates();
-
- boolean found = false;
- TimeOut timeOut = new TimeOut(5, TimeUnit.SECONDS, TimeSource.NANO_TIME);
- while (!timeOut.hasTimedOut()) {
- DocCollection c1 = reader.getClusterState().getCollection("c1");
- if ("y".equals(c1.getStr("x"))) {
- found = true;
- break;
- }
- }
- assertTrue("Could not find updated property in collection c1 even after
5 seconds", found);
- } finally {
- IOUtils.close(reader, zkClient);
- server.shutdown();
+ @After
+ public void tearDown() throws Exception {
+ if (fixture != null) {
+ fixture.close();
}
+ super.tearDown();
}
- public void testWatchedCollectionCreation() throws Exception {
- Path zkDir = createTempDir("testWatchedCollectionCreation");
-
+ private static TestFixture setupTestFixture(String testPrefix) throws
Exception {
+ Path zkDir = createTempDir(testPrefix);
ZkTestServer server = new ZkTestServer(zkDir);
+ server.run();
+ SolrZkClient zkClient =
+ new SolrZkClient(server.getZkAddress(),
OverseerTest.DEFAULT_CONNECTION_TIMEOUT);
+ ZkController.createClusterZkNodes(zkClient);
- SolrZkClient zkClient = null;
- ZkStateReader reader = null;
-
- try {
- server.run();
+ ZkStateReader reader = new ZkStateReader(zkClient);
+ reader.createClusterStateWatchersAndUpdate();
- zkClient = new SolrZkClient(server.getZkAddress(),
OverseerTest.DEFAULT_CONNECTION_TIMEOUT);
- ZkController.createClusterZkNodes(zkClient);
+ ZkStateWriter writer = new ZkStateWriter(reader, new Stats());
- reader = new ZkStateReader(zkClient);
- reader.createClusterStateWatchersAndUpdate();
- reader.registerCore("c1");
+ return new TestFixture(server, zkClient, reader, writer);
+ }
- // Initially there should be no c1 collection.
- assertNull(reader.getClusterState().getCollectionRef("c1"));
+ public void testExternalCollectionWatchedNotWatched() throws Exception {
+ ZkStateWriter writer = fixture.writer;
+ ZkStateReader reader = fixture.reader;
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
+
+ // create new collection
+ ZkWriteCommand c1 =
+ new ZkWriteCommand(
+ "c1",
+ new DocCollection(
+ "c1",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0));
+
+ writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(c1), null);
+ writer.writePendingUpdates();
+ reader.forceUpdateCollection("c1");
+
+
assertTrue(reader.getClusterState().getCollectionRef("c1").isLazilyLoaded());
+ reader.registerCore("c1");
+
assertFalse(reader.getClusterState().getCollectionRef("c1").isLazilyLoaded());
+ reader.unregisterCore("c1");
+
assertTrue(reader.getClusterState().getCollectionRef("c1").isLazilyLoaded());
+ }
- zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
- reader.forceUpdateCollection("c1");
+ public void testCollectionStateWatcherCaching() throws Exception {
+ ZkStateWriter writer = fixture.writer;
+ ZkStateReader reader = fixture.reader;
+
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
+
+ DocCollection state =
+ new DocCollection(
+ "c1",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0);
+ ZkWriteCommand wc = new ZkWriteCommand("c1", state);
+ writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
+ writer.writePendingUpdates();
+ assertTrue(fixture.zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE +
"/c1/state.json", true));
+ reader.waitForState(
+ "c1", 1, TimeUnit.SECONDS, (liveNodes, collectionState) ->
collectionState != null);
+
+ Map<String, Object> props = new HashMap<>();
+ props.put("x", "y");
+ props.put(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME);
+ state = new DocCollection("c1", new HashMap<>(), props, DocRouter.DEFAULT,
0);
+ wc = new ZkWriteCommand("c1", state);
+ writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
+ writer.writePendingUpdates();
+
+ boolean found = false;
+ TimeOut timeOut = new TimeOut(5, TimeUnit.SECONDS, TimeSource.NANO_TIME);
+ while (!timeOut.hasTimedOut()) {
+ DocCollection c1 = reader.getClusterState().getCollection("c1");
+ if ("y".equals(c1.getStr("x"))) {
+ found = true;
+ break;
+ }
+ }
+ assertTrue("Could not find updated property in collection c1 even after 5
seconds", found);
+ }
- // Still no c1 collection, despite a collection path.
- assertNull(reader.getClusterState().getCollectionRef("c1"));
+ public void testWatchedCollectionCreation() throws Exception {
+ ZkStateWriter writer = fixture.writer;
+ ZkStateReader reader = fixture.reader;
+
+ reader.registerCore("c1");
+
+ // Initially there should be no c1 collection.
+ assertNull(reader.getClusterState().getCollectionRef("c1"));
+
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
+ reader.forceUpdateCollection("c1");
+
+ // Still no c1 collection, despite a collection path.
+ assertNull(reader.getClusterState().getCollectionRef("c1"));
+
+ // create new collection
+ DocCollection state =
+ new DocCollection(
+ "c1",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0);
+ ZkWriteCommand wc = new ZkWriteCommand("c1", state);
+ writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
+ writer.writePendingUpdates();
+
+ assertTrue(fixture.zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE +
"/c1/state.json", true));
+
+ // reader.forceUpdateCollection("c1");
+ reader.waitForState("c1", TIMEOUT, TimeUnit.SECONDS, (n, c) -> c != null);
+ ClusterState.CollectionRef ref =
reader.getClusterState().getCollectionRef("c1");
+ assertNotNull(ref);
+ assertFalse(ref.isLazilyLoaded());
+ }
- ZkStateWriter writer = new ZkStateWriter(reader, new Stats());
+ public void testForciblyRefreshAllClusterState() throws Exception {
+ ZkStateWriter writer = fixture.writer;
+ ZkStateReader reader = fixture.reader;
+
+ reader.registerCore("c1"); // watching c1, so it should get non lazy
reference
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
+
+ reader.forciblyRefreshAllClusterStateSlow();
+ // Initially there should be no c1 collection.
+ assertNull(reader.getClusterState().getCollectionRef("c1"));
+
+ // create new collection
+ DocCollection state =
+ new DocCollection(
+ "c1",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0);
+ ZkWriteCommand wc = new ZkWriteCommand("c1", state);
+ writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
+ writer.writePendingUpdates();
+
+ assertTrue(fixture.zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE +
"/c1/state.json", true));
+
+ reader.forciblyRefreshAllClusterStateSlow();
+ ClusterState.CollectionRef ref =
reader.getClusterState().getCollectionRef("c1");
+ assertNotNull(ref);
+ assertFalse(ref.isLazilyLoaded());
+ assertEquals(0, ref.get().getZNodeVersion());
+
+ // update the collection
+ state =
+ new DocCollection(
+ "c1",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ ref.get().getZNodeVersion());
+ wc = new ZkWriteCommand("c1", state);
+ writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
+ writer.writePendingUpdates();
+
+ reader.forciblyRefreshAllClusterStateSlow();
+ ref = reader.getClusterState().getCollectionRef("c1");
+ assertNotNull(ref);
+ assertFalse(ref.isLazilyLoaded());
+ assertEquals(1, ref.get().getZNodeVersion());
+
+ // delete the collection c1, add a collection c2 that is NOT watched
+ ZkWriteCommand wc1 = new ZkWriteCommand("c1", null);
+
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c2", true);
+ state =
+ new DocCollection(
+ "c2",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0);
+ ZkWriteCommand wc2 = new ZkWriteCommand("c2", state);
+
+ writer.enqueueUpdate(reader.getClusterState(), Arrays.asList(wc1, wc2),
null);
+ writer.writePendingUpdates();
+
+ reader.forciblyRefreshAllClusterStateSlow();
+ ref = reader.getClusterState().getCollectionRef("c1");
+ assertNull(ref);
+
+ ref = reader.getClusterState().getCollectionRef("c2");
+ assertNotNull(ref);
+ assertTrue(
+ "c2 should have been lazily loaded but is not!",
+ ref.isLazilyLoaded()); // c2 should be lazily loaded as it's not
watched
+ assertEquals(0, ref.get().getZNodeVersion());
+ }
- // create new collection
- DocCollection state =
- new DocCollection(
- "c1",
- new HashMap<>(),
- Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
- DocRouter.DEFAULT,
- 0);
- ZkWriteCommand wc = new ZkWriteCommand("c1", state);
- writer.enqueueUpdate(reader.getClusterState(),
Collections.singletonList(wc), null);
- writer.writePendingUpdates();
+ public void testGetCurrentCollections() throws Exception {
+ ZkStateWriter writer = fixture.writer;
+ ZkStateReader reader = fixture.reader;
+
+ reader.registerCore("c1"); // listen to c1. not yet exist
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c1", true);
+ reader.forceUpdateCollection("c1");
+ Set<String> currentCollections = reader.getCurrentCollections();
+ assertEquals(0, currentCollections.size()); // no active collections yet
+
+ // now create both c1 (watched) and c2 (not watched)
+ DocCollection state1 =
+ new DocCollection(
+ "c1",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0);
+ ZkWriteCommand wc1 = new ZkWriteCommand("c1", state1);
+ DocCollection state2 =
+ new DocCollection(
+ "c2",
+ new HashMap<>(),
+ Map.of(ZkStateReader.CONFIGNAME_PROP,
ConfigSetsHandler.DEFAULT_CONFIGSET_NAME),
+ DocRouter.DEFAULT,
+ 0);
+
+ // do not listen to c2
+ fixture.zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/c2", true);
+ ZkWriteCommand wc2 = new ZkWriteCommand("c2", state2);
+
+ writer.enqueueUpdate(reader.getClusterState(), Arrays.asList(wc1, wc2),
null);
+ writer.writePendingUpdates();
+
+ reader.forceUpdateCollection("c1");
+ reader.forceUpdateCollection("c2");
+ currentCollections =
+ reader.getCurrentCollections(); // should detect both collections (c1
watched, c2 lazy
+ // loaded)
Review Comment:
Thanks! I will try again, I ran both `./gradlew :solr:solrj:spotlessApply`
and `./gradlew :solr:core:spotlessApply` and sometimes it produces some weird
wrapping...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@solr.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@solr.apache.org
For additional commands, e-mail: issues-h...@solr.apache.org
