SOLR-12955: Refactored DistributedUpdateProcessor to put SolrCloud specifics into a subclass

Closes #528

solr/CHANGES.txt

@@ -128,6 +128,7 @@ Bug Fixes
 
 Improvements
 ----------------------
+
 * SOLR-12999: Index replication could delete segments before downloading segments from master if there is not enough
   disk space (noble)
 
@@ -166,6 +167,9 @@ Other Changes
 
 * SOLR-8033: Remove debug if branch in HdfsTransactionLog (Kevin Risden)
 
+* SOLR-12955: Refactored DistributedUpdateProcessor to put SolrCloud functionality into a subclass.
+  (Bar Rotstein, David Smiley)
+
 ==================  8.0.0 ==================
 
 Consult the LUCENE_CHANGES.txt file for additional, low level, changes in this release.

solr/core/src/java/org/apache/solr/update/processor/CdcrUpdateProcessor.java

@@ -37,7 +37,7 @@ import org.slf4j.LoggerFactory;
  * by the target cluster.
  * </p>
  */
-public class CdcrUpdateProcessor extends DistributedUpdateProcessor {
+public class CdcrUpdateProcessor extends DistributedZkUpdateProcessor {
 
   public static final String CDCR_UPDATE = "cdcr.update";
 

solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessorFactory.java

@@ -49,9 +49,16 @@ public class DistributedUpdateProcessorFactory
   @Override
   public UpdateRequestProcessor getInstance(SolrQueryRequest req,
       SolrQueryResponse rsp, UpdateRequestProcessor next) {
+
+    final boolean isZkAware = req.getCore().getCoreContainer().isZooKeeperAware();
+
+    DistributedUpdateProcessor distribUpdateProcessor =
+        isZkAware ?
+            new DistributedZkUpdateProcessor(req, rsp, next) :
+            new DistributedUpdateProcessor(req, rsp, next);
     // note: will sometimes return DURP (no overhead) instead of wrapping
     return RoutedAliasUpdateProcessor.wrap(req,
-        new DistributedUpdateProcessor(req, rsp, next));
+        distribUpdateProcessor);
   }
   
 }

solr/core/src/java/org/apache/solr/update/processor/DocBasedVersionConstraintsProcessor.java

@@ -368,7 +368,8 @@ public class DocBasedVersionConstraintsProcessor extends UpdateRequestProcessor
       return true;
     }
     // if phase==TOLEADER, we can't just assume we are the leader... let the normal logic check.
-    return !distribProc.isLeader(cmd);
+    distribProc.setupRequest(cmd);
+    return !distribProc.isLeader();
   }
 
   @Override

solr/core/src/java/org/apache/solr/update/processor/SkipExistingDocumentsProcessorFactory.java

@@ -216,7 +216,8 @@ public class SkipExistingDocumentsProcessorFactory extends UpdateRequestProcesso
       if (phase == DistributedUpdateProcessor.DistribPhase.FROMLEADER) {
         return false;
       }
-      return distribProc.isLeader(cmd);
+      distribProc.setupRequest(cmd);
+      return distribProc.isLeader();
     }
 
     @Override

solr/core/src/test/org/apache/solr/update/processor/AtomicUpdateProcessorFactoryTest.java

@@ -232,7 +232,7 @@ public class AtomicUpdateProcessorFactoryTest extends SolrTestCaseJ4 {
 
           try {
             factory.getInstance(cmd.getReq(), new SolrQueryResponse(),
-                new DistributedUpdateProcessor(cmd.getReq(), new SolrQueryResponse(),
+                createDistributedUpdateProcessor(cmd.getReq(), new SolrQueryResponse(),
                     new RunUpdateProcessor(cmd.getReq(), null))).processAdd(cmd);
           } catch (IOException e) {
           }

solr/core/src/test/org/apache/solr/update/processor/DistributedUpdateProcessorTest.java

@@ -34,11 +34,10 @@ public class DistributedUpdateProcessorTest extends SolrTestCaseJ4 {
   }
 
   @Test
-  public void testShouldBufferUpdate() {
+  public void testShouldBufferUpdateZk() {
     SolrQueryRequest req = new LocalSolrQueryRequest(h.getCore(), new ModifiableSolrParams());
     DistributedUpdateProcessor processor = new DistributedUpdateProcessor(
         req, null, null, null);
-
     AddUpdateCommand cmd = new AddUpdateCommand(req);
     // applying buffer updates, isReplayOrPeerSync flag doesn't matter
     assertFalse(processor.shouldBufferUpdate(cmd, false, UpdateLog.State.APPLYING_BUFFERED));
@@ -50,5 +49,4 @@ public class DistributedUpdateProcessorTest extends SolrTestCaseJ4 {
     assertTrue(processor.shouldBufferUpdate(cmd, false, UpdateLog.State.APPLYING_BUFFERED));
   }
 
-
 }

solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java

@@ -131,6 +131,9 @@ import org.apache.solr.schema.IndexSchema;
 import org.apache.solr.schema.SchemaField;
 import org.apache.solr.search.SolrIndexSearcher;
 import org.apache.solr.servlet.DirectSolrConnection;
+import org.apache.solr.update.processor.DistributedUpdateProcessor;
+import org.apache.solr.update.processor.DistributedZkUpdateProcessor;
+import org.apache.solr.update.processor.UpdateRequestProcessor;
 import org.apache.solr.util.LogLevel;
 import org.apache.solr.util.RandomizeSSL;
 import org.apache.solr.util.RandomizeSSL.SSLRandomizer;
@@ -2892,6 +2895,14 @@ public abstract class SolrTestCaseJ4 extends SolrTestCase {
     }
   }
 
+  public static DistributedUpdateProcessor createDistributedUpdateProcessor(SolrQueryRequest req, SolrQueryResponse rsp,
+                                                                            UpdateRequestProcessor next) {
+    if(h.getCoreContainer().isZooKeeperAware()) {
+      return new DistributedZkUpdateProcessor(req, rsp, next);
+    }
+    return new DistributedUpdateProcessor(req, rsp, next);
+  }
+  
   /**
    * Cleans up the randomized sysproperties and variables set by {@link #randomizeNumericTypesProperties}
    *