You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by yo...@apache.org on 2019/10/30 20:45:06 UTC
[lucene-solr] 01/11: @W-6475448 Allow CorePullerFeeder to be provisioned with a callback in tests and… (#386)
This is an automated email from the ASF dual-hosted git repository.
yonik pushed a commit to branch jira/SOLR-13101
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit c7ad68fd304e6846d35c4af0d7e5d575233a1c46
Author: Andy Vuong <a....@salesforce.com>
AuthorDate: Thu Sep 19 21:05:43 2019 -0400
@W-6475448 Allow CorePullerFeeder to be provisioned with a callback in tests and… (#386)
* Allow CorePullerFeeder to be provisioned with a callback in tests and allow testing async pulls
---
.../solr/store/blob/process/BlobProcessUtil.java | 18 +++++--
.../solr/store/blob/process/CorePullTask.java | 2 +-
.../solr/store/blob/process/CorePullerFeeder.java | 6 ++-
.../solr/store/shared/SharedStoreManager.java | 8 +++
.../shared/SimpleSharedStoreEndToEndPullTest.java | 58 +++++++++++++++++++---
.../store/shared/SolrCloudSharedStoreTestCase.java | 19 +++++--
6 files changed, 93 insertions(+), 18 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/store/blob/process/BlobProcessUtil.java b/solr/core/src/java/org/apache/solr/store/blob/process/BlobProcessUtil.java
index d090952..164818d 100644
--- a/solr/core/src/java/org/apache/solr/store/blob/process/BlobProcessUtil.java
+++ b/solr/core/src/java/org/apache/solr/store/blob/process/BlobProcessUtil.java
@@ -30,9 +30,18 @@ public class BlobProcessUtil {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private volatile CorePullerFeeder runningFeeder = null;
+ /*
+ * Start the Blob store async core pull machinery
+ */
public BlobProcessUtil(CoreContainer coreContainer) {
- // Start the Blob store sync core push and async core pull machinery
- runningFeeder = initializeCorePullerFeeder(coreContainer);
+ this(coreContainer, new CorePullerFeeder(coreContainer));
+ }
+
+ /*
+ * Start the Blob store async core pull machinery
+ */
+ public BlobProcessUtil(CoreContainer coreContainer, CorePullerFeeder cpf) {
+ runningFeeder = initializeCorePullerFeeder(cpf);
}
/**
@@ -44,11 +53,10 @@ public class BlobProcessUtil {
/**
* Initializes the CorePullerFeeder and starts running thread
- * @param cores CoreContainer
+ * @param cpf CorePullerFeeder
* @return CorePullerFeeder
*/
- private CorePullerFeeder initializeCorePullerFeeder(CoreContainer cores) {
- CorePullerFeeder cpf = new CorePullerFeeder(cores);
+ private CorePullerFeeder initializeCorePullerFeeder(CorePullerFeeder cpf) {
Thread t = new Thread(cpf);
t.setName("blobPullerFeeder-" + t.getName());
t.start();
diff --git a/solr/core/src/java/org/apache/solr/store/blob/process/CorePullTask.java b/solr/core/src/java/org/apache/solr/store/blob/process/CorePullTask.java
index 2ba1f6c..e8af082 100644
--- a/solr/core/src/java/org/apache/solr/store/blob/process/CorePullTask.java
+++ b/solr/core/src/java/org/apache/solr/store/blob/process/CorePullTask.java
@@ -446,7 +446,7 @@ public class CorePullTask implements DeduplicatingList.Deduplicatable<String> {
* @param message
* Human readable message explaining a failure, or <code>null</code> if no message available.
*/
- void finishedPull(CorePullTask pullTask, BlobCoreMetadata blobMetadata, CoreSyncStatus status, String message)
+ public void finishedPull(CorePullTask pullTask, BlobCoreMetadata blobMetadata, CoreSyncStatus status, String message)
throws InterruptedException;
}
}
diff --git a/solr/core/src/java/org/apache/solr/store/blob/process/CorePullerFeeder.java b/solr/core/src/java/org/apache/solr/store/blob/process/CorePullerFeeder.java
index 4cae45d..b950d67 100644
--- a/solr/core/src/java/org/apache/solr/store/blob/process/CorePullerFeeder.java
+++ b/solr/core/src/java/org/apache/solr/store/blob/process/CorePullerFeeder.java
@@ -57,6 +57,10 @@ public class CorePullerFeeder extends CoreSyncFeeder {
String getMonsterThreadName() {
return PULLER_THREAD_PREFIX;
}
+
+ protected CorePullTask.PullCoreCallback getCorePullTaskCallback() {
+ return callback;
+ }
@Override
void feedTheMonsters() throws InterruptedException {
@@ -69,7 +73,7 @@ public class CorePullerFeeder extends CoreSyncFeeder {
PullCoreInfo pci = tracker.getCoreToPull();
// Add the core to the list consumed by the thread doing the actual work
- CorePullTask pt = new CorePullTask(cores, pci, callback);
+ CorePullTask pt = new CorePullTask(cores, pci, getCorePullTaskCallback());
pullTaskQueue.addDeduplicated(pt, /* isReenqueue */ false);
syncsEnqueuedSinceLastLog++;
diff --git a/solr/core/src/java/org/apache/solr/store/shared/SharedStoreManager.java b/solr/core/src/java/org/apache/solr/store/shared/SharedStoreManager.java
index 530cbef..3dd525a 100644
--- a/solr/core/src/java/org/apache/solr/store/shared/SharedStoreManager.java
+++ b/solr/core/src/java/org/apache/solr/store/shared/SharedStoreManager.java
@@ -50,6 +50,14 @@ public class SharedStoreManager {
this.blobStorageProvider = blobStorageProvider;
}
+ @VisibleForTesting
+ public void initBlobProcessUtil(BlobProcessUtil processUtil) {
+ if (blobProcessUtil != null) {
+ blobProcessUtil.shutdown();
+ }
+ blobProcessUtil = processUtil;
+ }
+
/*
* Initiates a SharedShardMetadataController if it doesn't exist and returns one
*/
diff --git a/solr/core/src/test/org/apache/solr/store/shared/SimpleSharedStoreEndToEndPullTest.java b/solr/core/src/test/org/apache/solr/store/shared/SimpleSharedStoreEndToEndPullTest.java
index 4e660f6..4a7921e 100644
--- a/solr/core/src/test/org/apache/solr/store/shared/SimpleSharedStoreEndToEndPullTest.java
+++ b/solr/core/src/test/org/apache/solr/store/shared/SimpleSharedStoreEndToEndPullTest.java
@@ -17,9 +17,14 @@
package org.apache.solr.store.shared;
import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
@@ -28,7 +33,13 @@ import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.SolrCore;
+import org.apache.solr.store.blob.client.BlobCoreMetadata;
import org.apache.solr.store.blob.client.CoreStorageClient;
+import org.apache.solr.store.blob.process.BlobProcessUtil;
+import org.apache.solr.store.blob.process.CorePullTask;
+import org.apache.solr.store.blob.process.CorePullTask.PullCoreCallback;
+import org.apache.solr.store.blob.process.CorePullerFeeder;
+import org.apache.solr.store.blob.process.CoreSyncStatus;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -64,11 +75,23 @@ public class SimpleSharedStoreEndToEndPullTest extends SolrCloudSharedStoreTestC
setupCluster(2);
CloudSolrClient cloudClient = cluster.getSolrClient();
- // setup the test harness
+ // setup the test harnesses
+ Map<String, CountDownLatch> cdlMap = new HashMap<>();
+
+ CountDownLatch latch1 = new CountDownLatch(1);
+ JettySolrRunner solrProcess1 = cluster.getJettySolrRunner(0);
CoreStorageClient storageClient1 = setupLocalBlobStoreClient(sharedStoreRootPath, DEFAULT_BLOB_DIR_NAME);
- setupTestSharedClientForNode(getBlobStorageProviderTestInstance(storageClient1), cluster.getJettySolrRunner(0));
+ setupTestSharedClientForNode(getBlobStorageProviderTestInstance(storageClient1), solrProcess1);
+ configureTestBlobProcessForNode(solrProcess1, setupCallback(latch1));
+
+ CountDownLatch latch2 = new CountDownLatch(1);
+ JettySolrRunner solrProcess2 = cluster.getJettySolrRunner(1);
CoreStorageClient storageClient2 = setupLocalBlobStoreClient(sharedStoreRootPath, DEFAULT_BLOB_DIR_NAME);
- setupTestSharedClientForNode(getBlobStorageProviderTestInstance(storageClient2), cluster.getJettySolrRunner(1));
+ setupTestSharedClientForNode(getBlobStorageProviderTestInstance(storageClient2), solrProcess2);
+ configureTestBlobProcessForNode(solrProcess2, setupCallback(latch2));
+
+ cdlMap.put(solrProcess1.getNodeName(), latch1);
+ cdlMap.put(solrProcess2.getNodeName(), latch2);
String collectionName = "sharedCollection";
int maxShardsPerNode = 1;
@@ -121,9 +144,9 @@ public class SimpleSharedStoreEndToEndPullTest extends SolrCloudSharedStoreTestC
resp = followerDirectClient.query(params);
assertEquals(0, resp.getResults().getNumFound());
- // TODO super ugly and inappropriate but the pull shouldn't take long. At some point we'll
- // make our end-to-end async testing nicer by supporting test listeners for the async tasks
- Thread.sleep(5000);
+ // wait until pull is finished
+ CountDownLatch latch = cdlMap.get(followerReplica.getNodeName());
+ assertTrue(latch.await(120, TimeUnit.SECONDS));
// do another query to verify we've pulled everything
resp = followerDirectClient.query(params);
@@ -140,4 +163,27 @@ public class SimpleSharedStoreEndToEndPullTest extends SolrCloudSharedStoreTestC
core.close();
}
}
+
+ private BlobProcessUtil configureTestBlobProcessForNode(JettySolrRunner runner, PullCoreCallback callback) {
+ CorePullerFeeder cpf = new CorePullerFeeder(runner.getCoreContainer()) {
+ @Override
+ protected CorePullTask.PullCoreCallback getCorePullTaskCallback() {
+ return callback;
+ }
+ };
+ BlobProcessUtil testUtil = new BlobProcessUtil(runner.getCoreContainer(), cpf);
+ setupTestBlobProcessUtilForNode(testUtil, runner);
+ return testUtil;
+ }
+
+ private PullCoreCallback setupCallback(CountDownLatch latch) {
+ return new PullCoreCallback() {
+ @Override
+ public void finishedPull(CorePullTask pullTask, BlobCoreMetadata blobMetadata, CoreSyncStatus status,
+ String message) throws InterruptedException {
+ assertTrue(status.isSuccess());
+ latch.countDown();
+ }
+ };
+ }
}
diff --git a/solr/core/src/test/org/apache/solr/store/shared/SolrCloudSharedStoreTestCase.java b/solr/core/src/test/org/apache/solr/store/shared/SolrCloudSharedStoreTestCase.java
index 0e7749f..738ff3f 100644
--- a/solr/core/src/test/org/apache/solr/store/shared/SolrCloudSharedStoreTestCase.java
+++ b/solr/core/src/test/org/apache/solr/store/shared/SolrCloudSharedStoreTestCase.java
@@ -24,6 +24,7 @@ import org.apache.solr.cloud.MiniSolrCloudCluster;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.store.blob.client.CoreStorageClient;
import org.apache.solr.store.blob.client.LocalStorageClient;
+import org.apache.solr.store.blob.process.BlobProcessUtil;
import org.apache.solr.store.blob.provider.BlobStorageProvider;
/**
@@ -69,11 +70,19 @@ public class SolrCloudSharedStoreTestCase extends SolrCloudTestCase {
}
/**
- * adds the test harness to each solr process in the MiniSolrCluster
+ * Configures the Solr process with the given BlobStorageProvider
*/
- protected static void setupTestSharedClientForNode(BlobStorageProvider providerTestHarness, JettySolrRunner solrRunner) {
+ protected static void setupTestSharedClientForNode(BlobStorageProvider testBlobStorageProvider, JettySolrRunner solrRunner) {
SharedStoreManager manager = solrRunner.getCoreContainer().getSharedStoreManager();
- manager.initBlobStorageProvider(providerTestHarness);
+ manager.initBlobStorageProvider(testBlobStorageProvider);
+ }
+
+ /**
+ * Configures the Solr process with the given BlobProcessUtil
+ */
+ protected static void setupTestBlobProcessUtilForNode(BlobProcessUtil testBlobProcessUtil, JettySolrRunner solrRunner) {
+ SharedStoreManager manager = solrRunner.getCoreContainer().getSharedStoreManager();
+ manager.initBlobProcessUtil(testBlobProcessUtil);
}
/**
@@ -86,13 +95,13 @@ public class SolrCloudSharedStoreTestCase extends SolrCloudTestCase {
}
protected static BlobStorageProvider getBlobStorageProviderTestInstance(CoreStorageClient client) {
- BlobStorageProvider providerTestHarness = new BlobStorageProvider() {
+ BlobStorageProvider testBlobStorageProvider = new BlobStorageProvider() {
@Override
public CoreStorageClient getClient() {
return client;
}
};
- return providerTestHarness;
+ return testBlobStorageProvider;
}
}