SOLR-5754: StreamingSolrClients.getErrors() returns a snapshot copy instead of the live list

changelog/unreleased/SOLR-5754-streaming-clients-error-snapshot.yml

@@ -0,0 +1,7 @@
+title: "Hardening: StreamingSolrClients.getErrors() returns a snapshot copy instead of the live list, and the retry pass takes and clears the errors in one atomic drain under the list's lock"
+type: changed
+authors:
+  - name: Nick Shanin
+links:
+  - name: SOLR-5754
+    url: https://issues.apache.org/jira/browse/SOLR-5754

solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java

@@ -104,7 +104,7 @@ private void doRetriesIfNeeded() throws IOException {
     // NOTE: retries will be forwards to a single url
 
     List<SolrError> errors = new ArrayList<>(this.errors);
-    errors.addAll(clients.getErrors());
+    errors.addAll(clients.drainErrors());
     List<SolrError> resubmitList = new ArrayList<>();
 
     if (log.isInfoEnabled() && errors.size() > 0) {
@@ -164,7 +164,6 @@ private void doRetriesIfNeeded() throws IOException {
       }
     }
 
-    clients.clearErrors();
     this.errors.clear();
     for (SolrError err : resubmitList) {
       if (err.req.node instanceof ForwardNode) {

solr/core/src/java/org/apache/solr/update/StreamingSolrClients.java

@@ -48,7 +48,8 @@ public class StreamingSolrClients {
   private HttpJettySolrClient httpClient;
 
   private Map<String, ConcurrentUpdateBaseSolrClient> solrClients = new HashMap<>();
-  private List<SolrError> errors = Collections.synchronizedList(new ArrayList<>());
+  // written by the client runner threads, see ErrorReportingConcurrentUpdateSolrClient
+  final List<SolrError> errors = Collections.synchronizedList(new ArrayList<>());
 
   private ExecutorService updateExecutor;
 
@@ -57,8 +58,20 @@ public StreamingSolrClients(UpdateShardHandler updateShardHandler) {
     this.httpClient = updateShardHandler.getUpdateOnlyHttpClient();
   }
 
+  /** A snapshot of the errors reported so far; later errors do not show up in it. */
   public List<SolrError> getErrors() {
-    return errors;
+    synchronized (errors) {
+      return new ArrayList<>(errors);
+    }
+  }
+
+  /** Returns the errors reported so far and clears them, so no error can arrive in between. */
+  public List<SolrError> drainErrors() {
+    synchronized (errors) {
+      List<SolrError> drained = new ArrayList<>(errors);
+      errors.clear();
+      return drained;
+    }
   }
 
   public void clearErrors() {

solr/core/src/test/org/apache/solr/update/StreamingSolrClientsTest.java

@@ -0,0 +1,74 @@
+/*
+ * 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.update;
+
+import static org.apache.solr.SolrTestCaseJ4.assumeWorkingMockito;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+import org.apache.solr.SolrTestCase;
+import org.apache.solr.update.SolrCmdDistributor.SolrError;
+import org.junit.BeforeClass;
+
+public class StreamingSolrClientsTest extends SolrTestCase {
+
+  @BeforeClass
+  public static void ensureWorkingMockito() {
+    assumeWorkingMockito();
+  }
+
+  private static SolrError error(int statusCode) {
+    SolrError error = new SolrError();
+    error.statusCode = statusCode;
+    return error;
+  }
+
+  public void testGetErrorsIsASnapshot() {
+    StreamingSolrClients clients = new StreamingSolrClients(mock(UpdateShardHandler.class));
+    clients.errors.add(error(500));
+
+    List<SolrError> snapshot = clients.getErrors();
+    assertEquals(1, snapshot.size());
+
+    // an error reported by a runner thread after the snapshot was taken is not part of it
+    clients.errors.add(error(503));
+    assertEquals(1, snapshot.size());
+    assertEquals(2, clients.getErrors().size());
+
+    // clearing the clients does not empty a list handed out earlier
+    clients.clearErrors();
+    assertEquals(1, snapshot.size());
+    assertEquals(0, clients.getErrors().size());
+  }
+
+  public void testDrainErrorsReturnsAndClears() {
+    StreamingSolrClients clients = new StreamingSolrClients(mock(UpdateShardHandler.class));
+    clients.errors.add(error(500));
+    clients.errors.add(error(503));
+
+    List<SolrError> drained = clients.drainErrors();
+    assertEquals(2, drained.size());
+    assertEquals(500, drained.get(0).statusCode);
+    assertEquals(503, drained.get(1).statusCode);
+    assertTrue(clients.getErrors().isEmpty());
+
+    // an error that arrives after the drain is kept for the next round of retries
+    clients.errors.add(error(502));
+    assertEquals(1, clients.drainErrors().size());
+    assertTrue(clients.drainErrors().isEmpty());
+  }
+}