diff --git a/changelog/unreleased/solr-18409.yml b/changelog/unreleased/solr-18409.yml new file mode 100644 index 00000000000..9a2fc67cdb8 --- /dev/null +++ b/changelog/unreleased/solr-18409.yml @@ -0,0 +1,8 @@ +title: SOLR-18409 CrossDC Producer shouldMirror not whitelisted for distrib update requests + +type: fixed +authors: + - name: Andrzej Bialecki +links: + - name: SOLR-18409 + url: https://issues.apache.org/jira/browse/SOLR-18409 diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java index fe6f44501e7..787b3292d42 100644 --- a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java +++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java @@ -70,7 +70,7 @@ // asked for public class DistributedUpdateProcessor extends UpdateRequestProcessor { - static final String PARAM_WHITELIST_CTX_KEY = + public static final String PARAM_WHITELIST_CTX_KEY = DistributedUpdateProcessor.class + "PARAM_WHITELIST_CTX_KEY"; public static final String DISTRIB_FROM_SHARD = "distrib.from.shard"; public static final String DISTRIB_FROM_COLLECTION = "distrib.from.collection"; diff --git a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactory.java b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactory.java index b159d060fc0..d82c1053c11 100644 --- a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactory.java +++ b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactory.java @@ -30,6 +30,7 @@ import static org.apache.solr.update.processor.DistributedUpdateProcessor.DistribPhase; import static org.apache.solr.update.processor.DistributingUpdateProcessorFactory.DISTRIB_UPDATE_PARAM; +import com.google.common.annotations.VisibleForTesting; import java.io.IOException; import java.lang.invoke.MethodHandles; import java.util.HashMap; @@ -49,6 +50,7 @@ import org.apache.solr.crossdc.common.KafkaMirroringSink; import org.apache.solr.request.SolrQueryRequest; import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.update.processor.DistributedUpdateProcessorFactory; import org.apache.solr.update.processor.DocBasedVersionConstraintsProcessorFactory; import org.apache.solr.update.processor.UpdateRequestProcessor; import org.apache.solr.update.processor.UpdateRequestProcessorFactory; @@ -207,7 +209,7 @@ public void inform(SolrCore core) { // core.getResourceLoader().newInstance(RequestMirroringHandler.class.getName(), // KafkaRequestMirroringHandler.class); - conf = new KafkaCrossDcConf(properties); + setKafkaCrossDcConf(new KafkaCrossDcConf(properties)); KafkaMirroringSink sink = new KafkaMirroringSink(conf); @@ -215,7 +217,17 @@ public void inform(SolrCore core) { core.addCloseHook(new MyCloseHook(closer)); producerMetrics = new ProducerMetrics(core.getSolrMetricsContext().getChildContext(this), core); - mirroringHandler = new KafkaRequestMirroringHandler(sink); + setMirroringHandler(new KafkaRequestMirroringHandler(sink)); + } + + @VisibleForTesting + void setKafkaCrossDcConf(KafkaCrossDcConf conf) { + this.conf = conf; + } + + @VisibleForTesting + void setMirroringHandler(KafkaRequestMirroringHandler mirroringHandler) { + this.mirroringHandler = mirroringHandler; } @Override @@ -231,6 +243,10 @@ public UpdateRequestProcessor getInstance( throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "mirroringHandler is null"); } + // allow distributed forwarding to other replicas/shards + DistributedUpdateProcessorFactory.addParamToDistributedRequestWhitelist( + req, SERVER_SHOULD_MIRROR); + // Check if mirroring is disabled in request params, defaults to true boolean doMirroring = req.getParams().getBool(SERVER_SHOULD_MIRROR, true); boolean mirrorCommits = conf.getBool(MIRROR_COMMITS); diff --git a/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactoryTest.java b/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactoryTest.java new file mode 100644 index 00000000000..b202c6991fc --- /dev/null +++ b/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/update/processor/MirroringUpdateRequestProcessorFactoryTest.java @@ -0,0 +1,71 @@ +/* + * 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.solr.crossdc.update.processor; + +import static org.apache.solr.crossdc.update.processor.MirroringUpdateRequestProcessorFactory.SERVER_SHOULD_MIRROR; +import static org.apache.solr.update.processor.DistributedUpdateProcessor.PARAM_WHITELIST_CTX_KEY; +import static org.mockito.Mockito.mock; + +import java.util.HashMap; +import java.util.Set; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.core.SolrCore; +import org.apache.solr.crossdc.common.KafkaCrossDcConf; +import org.apache.solr.request.SolrQueryRequestBase; +import org.apache.solr.response.SolrQueryResponse; +import org.apache.solr.update.processor.UpdateRequestProcessor; +import org.junit.BeforeClass; +import org.junit.Test; + +public class MirroringUpdateRequestProcessorFactoryTest extends SolrTestCaseJ4 { + + @BeforeClass + public static void ensureWorkingMockito() { + assumeWorkingMockito(); + } + + /** + * getInstance() must whitelist SERVER_SHOULD_MIRROR on the request so that when the add/delete is + * subsequently forwarded to other replicas/shards by DistributedUpdateProcessor, the param is + * preserved instead of being stripped, which previously caused the receiving replica to re-decide + * (and potentially re-mirror) independently. See SOLR-18409. + */ + @Test + @SuppressWarnings("unchecked") + public void testShouldMirrorParamWhitelists() { + MirroringUpdateRequestProcessorFactory factory = new MirroringUpdateRequestProcessorFactory(); + factory.setMirroringHandler(mock(KafkaRequestMirroringHandler.class)); + factory.setKafkaCrossDcConf(new KafkaCrossDcConf(new HashMap<>())); + + ModifiableSolrParams params = new ModifiableSolrParams(); + params.set(SERVER_SHOULD_MIRROR, "false"); + SolrQueryRequestBase req = new SolrQueryRequestBase(mock(SolrCore.class), params) {}; + SolrQueryResponse rsp = new SolrQueryResponse(); + UpdateRequestProcessor next = mock(UpdateRequestProcessor.class); + + // this should add the SERVER_SHOULD_MIRROR param to the request context whitelist + factory.getInstance(req, rsp, next); + + Set whitelist = (Set) req.getContext().get(PARAM_WHITELIST_CTX_KEY); + assertNotNull( + "shouldMirror param must be added to the distributed request whitelist", whitelist); + assertTrue( + "shouldMirror param missing from distributed request whitelist", + whitelist.contains(SERVER_SHOULD_MIRROR)); + } +}