You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@solr.apache.org by ge...@apache.org on 2022/10/11 14:28:26 UTC
[solr-sandbox] branch crossdc-wip updated: Check batch-size against "max" before mirroring (#48)
This is an automated email from the ASF dual-hosted git repository.
gerlowskija pushed a commit to branch crossdc-wip
in repository https://gitbox.apache.org/repos/asf/solr-sandbox.git
The following commit(s) were added to refs/heads/crossdc-wip by this push:
new 302abcd Check batch-size against "max" before mirroring (#48)
302abcd is described below
commit 302abcdeaccd72e1f0d38565c1e59e98ee85c6ff
Author: Jason Gerlowski <ge...@apache.org>
AuthorDate: Tue Oct 11 10:28:20 2022 -0400
Check batch-size against "max" before mirroring (#48)
Prior to this commit, MirroringUpdateProcessor checked that individual
documents didn't exceed the mirroring batch-size limit on their own, but
no check existed to ensure that the entire list of docs doesn't exceed
the limit.
This commit adds that check (in MirroringUpdateProcessor.finish()). If
the configured max batch-size is exceeded, MUP will log out an error
(which includes the IDs of all documents in the batch).
---
.../update/processor/MirroringUpdateProcessor.java | 35 +++++++++++++++-------
.../MirroringUpdateRequestProcessorFactory.java | 4 +--
.../solr/crossdc/SolrAndKafkaIntegrationTest.java | 29 ++++++++++++++----
3 files changed, 50 insertions(+), 18 deletions(-)
diff --git a/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateProcessor.java b/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateProcessor.java
index 1ad8f33..74ee1eb 100644
--- a/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateProcessor.java
+++ b/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateProcessor.java
@@ -29,6 +29,7 @@ import java.util.Collection;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.apache.solr.common.SolrException.ErrorCode.SERVER_ERROR;
@@ -48,6 +49,7 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
* then submitted on finish().
*/
private UpdateRequest mirrorRequest;
+ private long mirrorRequestBytes;
private final SolrParams mirrorParams;
@@ -55,7 +57,7 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
* Controls whether docs exceeding the max-size (and thus cannot be mirrored) are indexed locally.
*/
private final boolean indexUnmirrorableDocs;
- private final long maxDocSizeBytes;
+ private final long maxMirroringBatchSizeBytes;
/**
@@ -70,14 +72,14 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
public MirroringUpdateProcessor(final UpdateRequestProcessor next, boolean doMirroring,
final boolean indexUnmirrorableDocs,
- final long maxDocSizeBytes,
+ final long maxMirroringBatchSizeBytes,
final SolrParams mirroredReqParams,
final DistributedUpdateProcessor.DistribPhase distribPhase,
final RequestMirroringHandler requestMirroringHandler) {
super(next);
this.doMirroring = doMirroring;
this.indexUnmirrorableDocs = indexUnmirrorableDocs;
- this.maxDocSizeBytes = maxDocSizeBytes;
+ this.maxMirroringBatchSizeBytes = maxMirroringBatchSizeBytes;
this.mirrorParams = mirroredReqParams;
this.distribPhase = distribPhase;
this.requestMirroringHandler = requestMirroringHandler;
@@ -90,6 +92,7 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
if (mirrorRequest == null) {
mirrorRequest = new UpdateRequest();
mirrorRequest.setParams(new ModifiableSolrParams(mirrorParams));
+ mirrorRequestBytes = 0L;
}
if (log.isDebugEnabled())
log.debug("createOrGetMirrorRequest={}",
@@ -101,9 +104,10 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
final SolrInputDocument doc = cmd.getSolrInputDocument().deepCopy();
doc.removeField(CommonParams.VERSION_FIELD); // strip internal doc version
- final boolean tooLargeForKafka = exceedsMaxDocSize(doc);
+ final long estimatedDocSizeInBytes = ObjectSizeEstimator.estimate(doc);
+ final boolean tooLargeForKafka = estimatedDocSizeInBytes > maxMirroringBatchSizeBytes;
if (tooLargeForKafka && !indexUnmirrorableDocs) {
- log.warn("Skipping indexing of doc {} as it exceeds the doc-size limit ({} bytes) and is unmirrorable.", cmd.getPrintableId(), maxDocSizeBytes);
+ log.warn("Skipping indexing of doc {} as it exceeds the doc-size limit ({} bytes) and is unmirrorable.", cmd.getPrintableId(), maxMirroringBatchSizeBytes);
} else {
super.processAdd(cmd); // let this throw to prevent mirroring invalid reqs
}
@@ -112,9 +116,10 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
boolean isLeader = isLeader(cmd.getReq(), cmd.getIndexedIdStr(), null, cmd.getSolrInputDocument());
if (doMirroring && isLeader) {
if (tooLargeForKafka) {
- log.warn("Skipping mirroring of doc {} because estimated size exceeds batch size limit {} bytes", cmd.getPrintableId(), maxDocSizeBytes);
+ log.error("Skipping mirroring of doc {} because estimated size exceeds batch size limit {} bytes", cmd.getPrintableId(), maxMirroringBatchSizeBytes);
} else {
createAndOrGetMirrorRequest().add(doc, cmd.commitWithin, cmd.overwrite);
+ mirrorRequestBytes += estimatedDocSizeInBytes;
}
}
@@ -195,11 +200,6 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
}
}
- private boolean exceedsMaxDocSize(SolrInputDocument doc) {
- final long estimatedSizeInBytes = ObjectSizeEstimator.estimate(doc);
- return estimatedSizeInBytes > maxDocSizeBytes;
- }
-
private static void processDBQResults(SolrClient client, String collection, String uniqueField,
QueryResponse rsp)
throws SolrServerException, IOException {
@@ -258,6 +258,19 @@ public class MirroringUpdateProcessor extends UpdateRequestProcessor {
super.finish();
if (doMirroring && mirrorRequest != null) {
+ // We are configured to mirror, but short-circuit on batches we already know will fail (because they cumulatively
+ // exceed the mirroring max-size)
+ if (mirrorRequestBytes > maxMirroringBatchSizeBytes) {
+ final String batchedIds = mirrorRequest.getDocuments().stream()
+ .map(doc -> doc.getField("id").getValue().toString())
+ .collect(Collectors.joining(", "));
+ log.warn("Mirroring skipped for request because batch size {} bytes exceeds limit {} bytes. IDs: {}",
+ mirrorRequestBytes, maxMirroringBatchSizeBytes, batchedIds);
+ mirrorRequest = null;
+ mirrorRequestBytes = 0L;
+ return;
+ }
+
try {
requestMirroringHandler.mirror(mirrorRequest);
mirrorRequest = null; // so we don't accidentally submit it again
diff --git a/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateRequestProcessorFactory.java b/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateRequestProcessorFactory.java
index d0ec65a..06f7dba 100644
--- a/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateRequestProcessorFactory.java
+++ b/crossdc-producer/src/main/java/org/apache/solr/update/processor/MirroringUpdateRequestProcessorFactory.java
@@ -220,7 +220,7 @@ public class MirroringUpdateRequestProcessorFactory extends UpdateRequestProcess
// Check if mirroring is disabled in request params, defaults to true
boolean doMirroring = req.getParams().getBool(SERVER_SHOULD_MIRROR, true);
- final long maxDocSizeBytes = conf.getInt(MAX_REQUEST_SIZE_BYTES);
+ final long maxMirroringBatchSizeBytes = conf.getInt(MAX_REQUEST_SIZE_BYTES);
ModifiableSolrParams mirroredParams = null;
if (doMirroring) {
@@ -251,7 +251,7 @@ public class MirroringUpdateRequestProcessorFactory extends UpdateRequestProcess
log.trace("Create MirroringUpdateProcessor with mirroredParams={}", mirroredParams);
}
- return new MirroringUpdateProcessor(next, doMirroring, indexUnmirrorableDocs, maxDocSizeBytes, mirroredParams,
+ return new MirroringUpdateProcessor(next, doMirroring, indexUnmirrorableDocs, maxMirroringBatchSizeBytes, mirroredParams,
DistribPhase.parseParam(req.getParams().get(DISTRIB_UPDATE_PARAM)), doMirroring ? mirroringHandler : null);
}
diff --git a/crossdc-producer/src/test/java/org/apache/solr/crossdc/SolrAndKafkaIntegrationTest.java b/crossdc-producer/src/test/java/org/apache/solr/crossdc/SolrAndKafkaIntegrationTest.java
index ec1e7c0..7cba50a 100644
--- a/crossdc-producer/src/test/java/org/apache/solr/crossdc/SolrAndKafkaIntegrationTest.java
+++ b/crossdc-producer/src/test/java/org/apache/solr/crossdc/SolrAndKafkaIntegrationTest.java
@@ -53,7 +53,8 @@ import static org.mockito.Mockito.spy;
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private static final int MAX_DOC_SIZE_BYTES = Integer.valueOf(DEFAULT_MAX_REQUEST_SIZE);
+ private static final int MAX_MIRROR_BATCH_SIZE_BYTES = Integer.valueOf(DEFAULT_MAX_REQUEST_SIZE);
+ private static final int MAX_DOC_SIZE_BYTES = MAX_MIRROR_BATCH_SIZE_BYTES;
static final String VERSION_FIELD = "_version_";
@@ -147,10 +148,10 @@ import static org.mockito.Mockito.spy;
@After
public void tearDown() throws Exception {
super.tearDown();
- solrCluster1.getSolrClient().deleteByQuery("*:*");
- solrCluster2.getSolrClient().deleteByQuery("*:*");
- solrCluster1.getSolrClient().commit();
- solrCluster2.getSolrClient().commit();
+ solrCluster1.getSolrClient().deleteByQuery(COLLECTION, "*:*");
+ solrCluster2.getSolrClient().deleteByQuery(COLLECTION, "*:*");
+ solrCluster1.getSolrClient().commit(COLLECTION);
+ solrCluster2.getSolrClient().commit(COLLECTION);
// Delete alternate collection in case it was created by any tests.
if (CollectionAdminRequest.listCollections(solrCluster1.getSolrClient()).contains(ALT_COLLECTION)) {
@@ -245,6 +246,24 @@ import static org.mockito.Mockito.spy;
assertClusterEventuallyHasDocs(cluster1Client, ALT_COLLECTION, "*:*", 2);
assertCluster2EventuallyHasDocs(ALT_COLLECTION, normalDocQuery, 1);
assertCluster2EventuallyHasDocs(ALT_COLLECTION, "*:*", 1);
+
+ // Index batch of docs that will exceed the max mirroring batch size cumulatively (but not individually)
+ // Batch consists of 100 docs each roughly 1/100th of the max-batch-size
+ docsToIndex.clear();
+ for (int i = 0; i < 100; i++) {
+ final SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "cumulativelyTooLarge-" + System.currentTimeMillis() + "-" + i);
+ doc.addField("cumulativelyTooLarge_b", "true");
+ doc.addField("text", new String(new byte[MAX_MIRROR_BATCH_SIZE_BYTES / 100]));
+ docsToIndex.add(doc);
+ }
+ cluster1Client.add(ALT_COLLECTION, docsToIndex);
+ cluster1Client.commit(ALT_COLLECTION);
+
+ final String cumulativelyTooLargeQuery = "cumulativelyTooLarge_b:true";
+ // Primary (but not secondary) should have 100 additional docs
+ assertClusterEventuallyHasDocs(cluster1Client, ALT_COLLECTION, cumulativelyTooLargeQuery, 100);
+ assertCluster2EventuallyHasDocs(ALT_COLLECTION, cumulativelyTooLargeQuery, 0);
}
private void assertCluster2EventuallyHasDocs(String collection, String query, int expectedNumDocs) throws Exception {