SOLR-18409: CrossDC Producer: shouldMirror not whitelisted for distrib update requests

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

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";

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,15 +209,25 @@ 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);
 
     Closer closer = new Closer(sink);
     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);

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<String> whitelist = (Set<String>) 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));
+  }
+}