diff --git a/changelog/unreleased/solr-3284-cusc-failed-docs.yml b/changelog/unreleased/solr-3284-cusc-failed-docs.yml new file mode 100644 index 000000000000..66849e93654b --- /dev/null +++ b/changelog/unreleased/solr-3284-cusc-failed-docs.yml @@ -0,0 +1,12 @@ +# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc + +title: > + ConcurrentUpdateSolrClient can now report which documents failed to reach the server. + Register a handler via the Builder's withErrorHandler(...) to recover the documents of a + failed batch, for example to route them to a retry queue or dead-letter topic. +type: added +authors: + - name: Serhiy Bzhezytskyy +links: + - name: SOLR-3284 + url: https://issues.apache.org/jira/browse/SOLR-3284 diff --git a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java index 7e8686f1a39c..458f57004b75 100644 --- a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java +++ b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClient.java @@ -22,6 +22,8 @@ import java.io.InputStream; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; import java.util.Locale; import java.util.Objects; import java.util.concurrent.TimeUnit; @@ -75,10 +77,10 @@ protected ConcurrentUpdateJettySolrClient(Builder builder) { } @Override - protected StreamingResponse doSendUpdateStream(ConcurrentUpdateBaseSolrClient.Update update) + protected SentStream doSendUpdateStream(ConcurrentUpdateBaseSolrClient.Update update) throws IOException, InterruptedException { - InputStreamResponseListener jettyListener; - try (OutStream out = initOutStream(basePath, update.request(), update.collection())) { + OutStream out = initOutStream(basePath, update.request(), update.collection()); + try (out) { ConcurrentUpdateBaseSolrClient.Update upd = update; while (upd != null) { UpdateRequest req = upd.request(); @@ -87,15 +89,17 @@ protected StreamingResponse doSendUpdateStream(ConcurrentUpdateBaseSolrClient.Up queue.add(upd); break; } - send(out, upd.request(), upd.collection()); + send(out, upd); out.flush(); notifyQueueAndRunnersIfEmptyQueue(); upd = queue.poll(pollQueueTimeMillis, TimeUnit.MILLISECONDS); } - jettyListener = out.getResponseListener(); } - return new JettyStreamingResponse(jettyListener); + return new SentStream( + new JettyStreamingResponse(out.getResponseListener()), + out.getDocIds(), + update.collection()); } private static class OutStream implements Closeable { @@ -104,6 +108,7 @@ private static class OutStream implements Closeable { private final OutputStreamRequestContent content; private final InputStreamResponseListener responseListener; private final boolean isXml; + private final List docIds = new ArrayList<>(); public OutStream( String origCollection, @@ -123,6 +128,10 @@ boolean belongToThisStream(SolrRequest solrRequest, String collection) { && Objects.equals(origCollection, collection); } + List getDocIds() { + return docIds; + } + public void write(byte[] b) throws IOException { this.content.getOutputStream().write(b); } @@ -210,8 +219,11 @@ private OutStream initOutStream(String baseUrl, UpdateRequest updateRequest, Str return outStream; } - private void send(OutStream outStream, SolrRequest req, String collection) throws IOException { - assert outStream.belongToThisStream(req, collection); + private void send(OutStream outStream, ConcurrentUpdateBaseSolrClient.Update update) + throws IOException { + UpdateRequest req = update.request(); + assert outStream.belongToThisStream(req, update.collection()); + outStream.docIds.addAll(idsForErrorReporting(req)); client.getRequestWriter().write(req, outStream.content.getOutputStream()); if (outStream.isXml) { // check for commit or optimize diff --git a/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java b/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java index df28d17f3b0d..10808dedba45 100644 --- a/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java +++ b/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/ConcurrentUpdateJettySolrClientTest.java @@ -43,6 +43,21 @@ public ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient( .build(); } + @Override + public ConcurrentUpdateBaseSolrClient errorHandlerConcurrentClient( + String serverUrl, + int queueSize, + int threadCount, + HttpSolrClient solrClient, + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler errorHandler) { + return new ConcurrentUpdateJettySolrClient.Builder(serverUrl, (HttpJettySolrClient) solrClient) + .withQueueSize(queueSize) + .withThreadCount(threadCount) + .setPollQueueTime(0, TimeUnit.MILLISECONDS) + .withErrorHandler(errorHandler) + .build(); + } + @Override public HttpSolrClient solrClient(Integer overrideIdleTimeoutMs) { var builder = new HttpJettySolrClient.Builder(); diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java index 752e7131509d..ce4c8673bb42 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateBaseSolrClient.java @@ -22,7 +22,9 @@ import java.lang.invoke.MethodHandles; import java.net.HttpURLConnection; import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -37,6 +39,7 @@ import org.apache.solr.client.solrj.request.UpdateRequest; import org.apache.solr.client.solrj.util.ClientUtils; import org.apache.solr.common.SolrException; +import org.apache.solr.common.SolrInputDocument; import org.apache.solr.common.params.SolrParams; import org.apache.solr.common.params.UpdateParams; import org.apache.solr.common.util.ExecutorUtil; @@ -71,6 +74,40 @@ public abstract class ConcurrentUpdateBaseSolrClient extends SolrClient { protected StallDetection stallDetection; + private final UpdateErrorHandler errorHandler; + + /** + * Callback invoked when a batch fails to reach the server. Register one via {@link + * Builder#withErrorHandler} to handle update errors comprehensively -- e.g. route the failed + * document ids to a retry queue or dead-letter topic. Implementations must be thread-safe; see + * {@link Builder#withErrorHandler} for the full threading contract. + */ + @FunctionalInterface + public interface UpdateErrorHandler { + /** + * Invoked when specific documents failed to reach the server. + * + * @param ex the error that occurred + * @param failedIds the ids of the documents that did not reach the server + * @param collection the collection the batch targeted, or null + */ + void onError(Throwable ex, List failedIds, String collection); + + /** + * Invoked for an error not tied to specific documents (e.g. a failure before any document was + * sent). Defaults to logging. + */ + default void onError(Throwable ex) { + LoggerFactory.getLogger(UpdateErrorHandler.class).error("update error", ex); + } + + /** The id used to report a failed document; defaults to its {@code id} field. */ + default String idOf(SolrInputDocument doc) { + Object id = doc.getFieldValue("id"); + return id == null ? null : id.toString(); + } + } + protected static class CustomBlockingQueue implements Iterable { private final BlockingQueue queue; private final Semaphore available; @@ -158,6 +195,7 @@ protected ConcurrentUpdateBaseSolrClient(Builder builder) { this.basePath = builder.baseSolrUrl; this.defaultCollection = builder.defaultCollection; this.pollQueueTimeMillis = builder.pollQueueTimeMillis; + this.errorHandler = builder.errorHandler; // Initialize stall detection long stallTimeMillis = Integer.getInteger("solr.cloud.client.stallTime", 15000); @@ -187,6 +225,31 @@ protected ConcurrentUpdateBaseSolrClient(Builder builder) { /** Class representing an UpdateRequest and an optional collection. */ protected record Update(UpdateRequest request, String collection) {} + /** + * The result of sending updates as a single stream: the response to await, plus the ids of the + * documents sent (for error reporting) and their collection. + */ + public record SentStream(StreamingResponse response, List docIds, String collection) {} + + /** + * The ids of a request's documents for error reporting, via {@link UpdateErrorHandler#idOf}. + * Empty when no error handler is registered, so the documents are not read and no ids are + * retained. + */ + protected List idsForErrorReporting(UpdateRequest request) { + if (errorHandler == null || request.getDocuments() == null) { + return List.of(); + } + List ids = new ArrayList<>(request.getDocuments().size()); + for (SolrInputDocument doc : request.getDocuments()) { + String id = errorHandler.idOf(doc); + if (id != null) { + ids.add(id); + } + } + return ids; + } + /** Opens a connection and sends everything... */ class Runner implements Runnable { @@ -235,17 +298,20 @@ void sendUpdateStream() throws Exception { try { while (!queue.isEmpty()) { InputStream rspBody = null; + List docIds = List.of(); + String collection = null; try { - Update update; notifyQueueAndRunnersIfEmptyQueue(); - update = queue.poll(pollQueueTimeMillis, TimeUnit.MILLISECONDS); + Update update = queue.poll(pollQueueTimeMillis, TimeUnit.MILLISECONDS); if (update == null) { break; } - StreamingResponse responseListener = null; - responseListener = doSendUpdateStream(update); + SentStream sent = doSendUpdateStream(update); + StreamingResponse responseListener = sent.response(); + docIds = sent.docIds(); + collection = sent.collection(); // just wait for the headers, so the idle timeout is sensible int statusCode = responseListener.awaitResponse(idleTimeoutMillis); @@ -266,12 +332,16 @@ void sendUpdateStream() throws Exception { solrExc = new RemoteSolrException(basePath, statusCode, remoteError); } - handleError(solrExc); + handleError(solrExc, docIds, collection); } else { onSuccess(responseListener.getUnderlyingResponse(), rspBody); } stallDetection.incrementProcessedCount(); + } catch (OutOfMemoryError | InterruptedException e) { + throw e; + } catch (Throwable e) { + handleError(e, docIds, collection); } finally { try { consumeFully(rspBody); @@ -287,7 +357,7 @@ void sendUpdateStream() throws Exception { } } - protected abstract StreamingResponse doSendUpdateStream(Update update) + protected abstract SentStream doSendUpdateStream(Update update) throws IOException, InterruptedException; private void consumeFully(InputStream is) { @@ -544,6 +614,22 @@ public void handleError(Throwable ex) { log.error("error", ex); } + private void handleError(Throwable ex, List failedIds, String collection) { + if (errorHandler == null) { + handleError(ex); + return; + } + try { + if (failedIds.isEmpty()) { + errorHandler.onError(ex); + } else { + errorHandler.onError(ex, failedIds, collection); + } + } catch (Exception handlerEx) { + log.error("errorHandler threw while handling a failed update", handlerEx); + } + } + /** * Intended to be used as an extension point for doing post-processing after a request completes. * @@ -627,6 +713,7 @@ public abstract static class Builder { protected boolean streamDeletes; protected boolean closeHttpClient; protected long pollQueueTimeMillis; + protected UpdateErrorHandler errorHandler; /** * Initialize a Builder object, based on the provided URL and client. @@ -763,6 +850,26 @@ public Builder setPollQueueTime(long pollQueueTime, TimeUnit unit) { return this; } + /** + * Registers a handler invoked when a batch fails to reach the server. It receives the ids of + * the failed documents (via {@link UpdateErrorHandler#idOf}, default the {@code id} field) and + * the target collection, so callers can recover them -- for example by routing the ids to a + * retry queue or dead-letter topic. + * + *

The handler is invoked on the client's internal runner threads and may be called + * concurrently when several batches fail at once, so implementations must be thread-safe. The + * call happens inline on a runner thread, so implementations should return quickly and not + * block, or they will slow update throughput. An exception thrown by the handler is logged and + * does not stop the client. + * + *

If no handler is registered the client falls back to its default behavior of logging the + * error, and document ids are not extracted. + */ + public Builder withErrorHandler(UpdateErrorHandler errorHandler) { + this.errorHandler = errorHandler; + return this; + } + /** * Create a {@link ConcurrentUpdateBaseSolrClient} based on the provided configuration options. */ diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java index 8ebbd64bd804..fc11824775d7 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClient.java @@ -36,41 +36,43 @@ protected ConcurrentUpdateJdkSolrClient(ConcurrentUpdateJdkSolrClient.Builder bu } @Override - protected StreamingResponse doSendUpdateStream(Update update) { + protected SentStream doSendUpdateStream(Update update) { UpdateRequest req = update.request(); String collection = update.collection(); CompletableFuture> resp = client.requestInputStreamAsync(basePath, req, collection); - return new StreamingResponse() { + StreamingResponse response = + new StreamingResponse() { - @Override - public int awaitResponse(long timeoutMillis) throws Exception { - return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode(); - } + @Override + public int awaitResponse(long timeoutMillis) throws Exception { + return resp.get(timeoutMillis, TimeUnit.MILLISECONDS).statusCode(); + } - @Override - public InputStream getInputStream() { - try { - return resp.get().body(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return InputStream.nullInputStream(); - } catch (ExecutionException e) { - throw new RuntimeException(e); - } - } + @Override + public InputStream getInputStream() { + try { + return resp.get().body(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return InputStream.nullInputStream(); + } catch (ExecutionException e) { + throw new RuntimeException(e); + } + } - @Override - public Object getUnderlyingResponse() { - return resp; - } + @Override + public Object getUnderlyingResponse() { + return resp; + } - @Override - public void close() throws IOException { - // No-op: InputStream is managed by java.net.http.HttpClient - } - }; + @Override + public void close() throws IOException { + // No-op: InputStream is managed by java.net.http.HttpClient + } + }; + return new SentStream(response, idsForErrorReporting(req), collection); } public static class Builder extends ConcurrentUpdateBaseSolrClient.Builder { diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java index 13e790b60b2b..9cfc922c975f 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java +++ b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateJdkSolrClientTest.java @@ -71,6 +71,21 @@ public ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient( .build(); } + @Override + public ConcurrentUpdateBaseSolrClient errorHandlerConcurrentClient( + String serverUrl, + int queueSize, + int threadCount, + HttpSolrClient solrClient, + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler errorHandler) { + return new ConcurrentUpdateJdkSolrClient.Builder(serverUrl, (HttpJdkSolrClient) solrClient) + .withQueueSize(queueSize) + .withThreadCount(threadCount) + .setPollQueueTime(0, TimeUnit.MILLISECONDS) + .withErrorHandler(errorHandler) + .build(); + } + public static class OutcomeCountingConcurrentUpdateSolrClient extends ConcurrentUpdateJdkSolrClient { private final AtomicInteger successCounter; diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java index 29e73a44007d..7a4c387e4083 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java +++ b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClientTestBase.java @@ -33,8 +33,10 @@ import java.net.http.HttpConnectTimeoutException; import java.util.Enumeration; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -82,6 +84,13 @@ public abstract ConcurrentUpdateBaseSolrClient outcomeCountingConcurrentClient( AtomicInteger failureCounter, StringBuilder errors); + public abstract ConcurrentUpdateBaseSolrClient errorHandlerConcurrentClient( + String serverUrl, + int queueSize, + int threadCount, + HttpSolrClient solrClient, + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler errorHandler); + /** Mock endpoint where the CUSS being tested in this class sends requests. */ public static class TestServlet extends HttpServlet implements JavaBinUpdateRequestCodec.StreamingUpdateHandler { @@ -273,6 +282,184 @@ public void testConcurrentUpdate() throws Exception { } } + /** + * Against a real Solr: a valid document succeeds and is not reported, while documents the server + * rejects (a non-numeric value for the pint "popularity" field) are reported by id -- so only the + * failed ids reach the handler. + */ + @Test + public void testFailedDocsAreRecoverableViaErrorHandler() throws Exception { + List failedIds = new CopyOnWriteArrayList<>(); + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient( + solrTestRule.getBaseUrl(), + 10, + 2, + httpClient, + (ex, ids, collection) -> failedIds.addAll(ids))) { + + SolrInputDocument good = new SolrInputDocument(); + good.addField("id", "good-1"); + concurrentClient.add("collection1", good); + concurrentClient.blockUntilFinished(); + assertTrue("a successful doc must not be reported: " + failedIds, failedIds.isEmpty()); + + for (int i = 1; i <= 5; i++) { + SolrInputDocument bad = new SolrInputDocument(); + bad.addField("id", "bad-" + i); + bad.addField("popularity", "not-an-int"); // rejected: pint field + concurrentClient.add("collection1", bad); + } + concurrentClient.blockUntilFinished(); + } + + assertEquals("only the 5 rejected docs should be reported: " + failedIds, 5, failedIds.size()); + for (int i = 1; i <= 5; i++) { + assertTrue("missing bad-" + i + " in " + failedIds, failedIds.contains("bad-" + i)); + } + assertFalse("the successful doc must not be reported", failedIds.contains("good-1")); + } + + /** Overriding idOf reports failures by a uniqueKey field other than "id". */ + @Test + public void testErrorHandlerWithCustomIdField() throws Exception { + TestServlet.clear(); + TestServlet.setErrorCode(500); // every request fails server-side + + String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo"; + List failedIds = new CopyOnWriteArrayList<>(); + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler handler = + new ConcurrentUpdateBaseSolrClient.UpdateErrorHandler() { + @Override + public void onError(Throwable ex, List ids, String collection) { + failedIds.addAll(ids); + } + + @Override + public String idOf(SolrInputDocument doc) { + Object v = doc.getFieldValue("record_uuid"); + return v == null ? null : v.toString(); + } + }; + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient(serverUrl, 10, 2, httpClient, handler)) { + + for (int i = 1; i <= 5; i++) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("id", "ignored-" + i); // not the uniqueKey the handler reads + doc.addField("record_uuid", "uuid-" + i); + concurrentClient.add("collection1", doc); + } + concurrentClient.blockUntilFinished(); + } + + assertEquals("all 5 docs should be reported by record_uuid", 5, failedIds.size()); + for (int i = 1; i <= 5; i++) { + assertTrue("missing uuid-" + i + " in " + failedIds, failedIds.contains("uuid-" + i)); + } + } + + /** + * An error not tied to identifiable documents (a send failure where the document has no + * resolvable id) reaches the general onError(Throwable) callback, not the per-id one. + */ + @Test + public void testGeneralErrorCallbackForNonDocumentError() throws Exception { + String unreachable = "http://localhost:1/solr"; // nothing listening -> connect failure + AtomicInteger richCalls = new AtomicInteger(); + AtomicInteger generalCalls = new AtomicInteger(); + ConcurrentUpdateBaseSolrClient.UpdateErrorHandler handler = + new ConcurrentUpdateBaseSolrClient.UpdateErrorHandler() { + @Override + public void onError(Throwable ex, List ids, String collection) { + richCalls.incrementAndGet(); + } + + @Override + public void onError(Throwable ex) { + generalCalls.incrementAndGet(); + } + }; + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient(unreachable, 10, 2, httpClient, handler)) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("name", "no id here"); // no id -> idOf returns null -> nothing to report per-id + concurrentClient.add("collection1", doc); + concurrentClient.blockUntilFinished(); + } + + assertTrue("general callback should fire for a non-document error", generalCalls.get() > 0); + assertEquals( + "the per-id callback should not fire for a non-document error", 0, richCalls.get()); + } + + /** + * A naive lambda handler only registers for per-id failures; an error not tied to identifiable + * documents is logged by the default general callback and does not stop the client. + */ + @Test + public void testNaiveLambdaSurvivesNonDocumentError() throws Exception { + String unreachable = "http://localhost:1/solr"; + List failedIds = new CopyOnWriteArrayList<>(); + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient( + unreachable, 10, 2, httpClient, (ex, ids, collection) -> failedIds.addAll(ids))) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("name", "no id here"); + concurrentClient.add("collection1", doc); + concurrentClient.blockUntilFinished(); // returns normally; default callback logged the error + } + + assertTrue( + "an error not tied to documents must not reach the per-id lambda: " + failedIds, + failedIds.isEmpty()); + } + + /** + * A handler that throws is contained and does not stop the client: every failed doc is still + * reported and blockUntilFinished() returns normally. + */ + @Test + public void testThrowingErrorHandlerDoesNotStopClient() throws Exception { + TestServlet.clear(); + TestServlet.setErrorCode(500); // every request fails server-side + + String serverUrl = solrTestRule.getBaseUrl() + "/cuss/foo"; + List seenIds = new CopyOnWriteArrayList<>(); + + try (var httpClient = solrClient(null); + var concurrentClient = + errorHandlerConcurrentClient( + serverUrl, + 10, + 2, + httpClient, + (ex, ids, collection) -> { + seenIds.addAll(ids); + throw new RuntimeException("handler blew up"); + })) { + + for (int i = 1; i <= 5; i++) { + SolrInputDocument doc = new SolrInputDocument(); + doc.addField("id", "doc-" + i); + concurrentClient.add("collection1", doc); + } + concurrentClient.blockUntilFinished(); + } + + for (int i = 1; i <= 5; i++) { + assertTrue("missing doc-" + i + " in " + seenIds, seenIds.contains("doc-" + i)); + } + } + @Test public void testCollectionParameters() throws IOException, SolrServerException {