diff --git a/changelog/unreleased/SOLR-5754-streaming-clients-error-snapshot.yml b/changelog/unreleased/SOLR-5754-streaming-clients-error-snapshot.yml new file mode 100644 index 000000000000..934ec30fb133 --- /dev/null +++ b/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 diff --git a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java index ea1d27897007..529ac44ac17d 100644 --- a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java +++ b/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 errors = new ArrayList<>(this.errors); - errors.addAll(clients.getErrors()); + errors.addAll(clients.drainErrors()); List 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) { diff --git a/solr/core/src/java/org/apache/solr/update/StreamingSolrClients.java b/solr/core/src/java/org/apache/solr/update/StreamingSolrClients.java index dc0b5d775567..64b37499fc91 100644 --- a/solr/core/src/java/org/apache/solr/update/StreamingSolrClients.java +++ b/solr/core/src/java/org/apache/solr/update/StreamingSolrClients.java @@ -48,7 +48,8 @@ public class StreamingSolrClients { private HttpJettySolrClient httpClient; private Map solrClients = new HashMap<>(); - private List errors = Collections.synchronizedList(new ArrayList<>()); + // written by the client runner threads, see ErrorReportingConcurrentUpdateSolrClient + final List 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 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 drainErrors() { + synchronized (errors) { + List drained = new ArrayList<>(errors); + errors.clear(); + return drained; + } } public void clearErrors() { diff --git a/solr/core/src/test/org/apache/solr/update/StreamingSolrClientsTest.java b/solr/core/src/test/org/apache/solr/update/StreamingSolrClientsTest.java new file mode 100644 index 000000000000..70b0878926d9 --- /dev/null +++ b/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 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 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()); + } +}