SOLR Cloud(7)Client Error Handler

SOLR Cloud(7)Client Error Handler

Achieved Replication Factor
Consider a collection with one shard and a replication factor of three. In this case, you have a shard leader and two additional replicas.The request is still considered successful from the perspective of the client.
Solr supports the optional min_rf parameter on update requests

If min_rf >=1, then Solo would return rf=1 in the response if the request only success on leader.

It is not supported in my Driver. https://github.com/inoio/solrs

It should be working in SOLRJ
http://lucene.472066.n3.nabble.com/SolrCloud-use-of-quot-min-rf-quot-through-SolrJ-td4164966.html

I may just try catch exceptions and make the current thread sleep for some time.
The core parts of the codes are as follow:

package com.sillycat.jobssolrconsumer.service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import org.apache.solr.client.solrj.response.UpdateResponse;
import org.apache.solr.common.SolrInputDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.amazonaws.services.sqs.AmazonSQSAsync;
import com.amazonaws.services.sqs.model.DeleteMessageBatchRequest;
import com.amazonaws.services.sqs.model.DeleteMessageBatchRequestEntry;
import com.amazonaws.services.sqs.model.DeleteMessageBatchResult;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.ReceiveMessageResult;
import com.sillycat.jobssolrconsumer.model.SqsFutureReceiver;
import com.sillycat.jobssolrconsumer.model.SqsMessagesSolrResponse;
import com.sillycat.jobssolrconsumer.util.MessageUtil;

import io.ino.solrs.JavaAsyncSolrClient;

public class ArchiveJobsToSolrTask implements Runnable {

    private static final int HALF_HOUR = 180;

    private static final long STARTTIME = System.currentTimeMillis();

    private static final Logger logger = LoggerFactory.getLogger(ArchiveJobsToSolrTask.class);

    private final AmazonSQSAsync sqsClient;
    private final JavaAsyncSolrClient solrClient;
    …snip...
    private final int TEN_SECOND_DURATION = 10 * 1000; // seconds
    private int repeatedFailures = 0;
   
    …snip...
    public void run() {
        try {
            while (true) {
                // https://en.wikipedia.org/wiki/Reactor_pattern
                this.inflightMsg.acquire();
                // this.rateLimiter.acquire();

                logger.trace("Worker {} - Request for more SQS messages", workerId);
                SqsFutureReceiver<ReceiveMessageRequest, ReceiveMessageResult> processSqsMessageFuture = new SqsFutureReceiver<>();
                sqsClient.receiveMessageAsync(receiveMessageRequest, processSqsMessageFuture);

                logger.trace("Worker {} - Block thread for response", workerId);
                ReceiveMessageResult response = processSqsMessageFuture.get();

                if (response.getMessages().size() > 0) {
                    logger.trace("Worker {} - Process response in a non-blocking fashion", workerId);
                    processSqsMessages(sqsClient, solrClient, response).thenAccept(batchMessageProcessedCount -> {
                        logger.trace("Update message count");
                        long totalMessageProcessed = messageCount.addAndGet(batchMessageProcessedCount);

                        // log rates
                        if (totalMessageProcessed % 100 == 0) {
                            long currentTime = System.currentTimeMillis();
                            long duration = (currentTime - STARTTIME) / 1000;
                            logger.info("Worker {} - Jobs count {}; {} jobs/second", workerId, totalMessageProcessed,
                                    totalMessageProcessed * 10 / duration);
                        }
                    }).whenComplete((value, ex) -> {
                        this.inflightMsg.release();
                    });
                } else {
                    // release the ticket, wait for another 10 seconds
                    this.inflightMsg.release();
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            logger.warn("Worker {} - Thread was interupted; terminating archiving SQS jobs to Solr.", workerId, e);
        } catch (Throwable t) {
            logger.warn("Worker {} - Worker threw an uncaught exception; terminating archiving SQS jobs to Solr.",
                    workerId, t);
            throw t;
        }
    }

    private CompletableFuture<Integer> processSqsMessages(AmazonSQSAsync sqsClient, JavaAsyncSolrClient solrClient,
            ReceiveMessageResult response) {
        return CompletableFuture.supplyAsync(() -> response)
                .thenCompose((receiveMessageResult) -> addSqsJobsToSolr(solrClient, receiveMessageResult))
                .thenCompose((messagesAndSolrResponse) -> deleteSuccessfullyProcessedMessagesFromSqs(sqsClient,
                        messagesAndSolrResponse))
                .thenApply((deleteResults) -> processSqsDeleteResults(deleteResults)).exceptionally((ex) -> {
                    logger.error("Unable to process SQS job messages", ex);
                    this.handleRetryPolicy();
                    return 0;
                });
    }

    private void handleRetryPolicy() {
        this.repeatedFailures = this.repeatedFailures + 1;
        try {
            // 1x1 = 1, 2x2 = 4, 3*3 = 9, 4*4 16, 5*5 = 15, 6*6 = 36
            int maxFailure = Math.min(this.repeatedFailures * this.repeatedFailures, HALF_HOUR);
            Thread.sleep(this.TEN_SECOND_DURATION * maxFailure);
        } catch (InterruptedException e) {
            logger.error("Thread Sleep Fail:", e);
        }
    }

    private CompletionStage<SqsMessagesSolrResponse> addSqsJobsToSolr(JavaAsyncSolrClient solrClient,
            ReceiveMessageResult receiveMessageResult) {
        logger.trace("Process the SQS message contents");
        final List<Message> messages = receiveMessageResult.getMessages();

        List<SolrInputDocument> solrJobs = null;
        try {
            solrJobs = convertSqsMessageToSolrJob(messages);
        } catch (Throwable t) {
            throw new RuntimeException("Unable to convert SQS message to Solr Jobs", t);
        }

        logger.trace("Send to Solr");
        CompletionStage<UpdateResponse> solrResponse = solrClient.addDocs(solrJobs);
        return solrResponse.thenApply((updateResponse) -> new SqsMessagesSolrResponse(messages, updateResponse));
    }

    private Integer processSqsDeleteResults(DeleteMessageBatchResult deleteResults) {
        if (deleteResults.getFailed().size() != 0) {
            logger.error("Unable to process {} messages", deleteResults.getFailed().size());
            throw new RuntimeException("Unable to process " + deleteResults.getFailed().size() + " messages");
        }
        // successful execute, reset the failure to 0
        this.repeatedFailures = 0;
        return deleteResults.getSuccessful().size();
    }

    private CompletionStage<DeleteMessageBatchResult> deleteSuccessfullyProcessedMessagesFromSqs(
            AmazonSQSAsync sqsClient, SqsMessagesSolrResponse messagesAndSolrResponse) {
        if (messagesAndSolrResponse.getUpdateResponse().getStatus() != 0) {
            throw new RuntimeException("Error response status of "
                    + messagesAndSolrResponse.getUpdateResponse().getStatus() + " from SOLR server");
        }

        List<Message> messages = messagesAndSolrResponse.getMessages();

        logger.trace("Return future of Solr success with original SQS Messages");

        // TODO consider always deleting messages that fail to process
        // TODO delete all messages that were processed
        final AtomicLong counter = new AtomicLong(0L);
        List<DeleteMessageBatchRequestEntry> deleteEntries = messages.stream()
                .map((m) -> new DeleteMessageBatchRequestEntry().withId("" + counter.getAndIncrement())
                        .withReceiptHandle(m.getReceiptHandle()))
                .collect(Collectors.toList());

        SqsFutureReceiver<DeleteMessageBatchRequest, DeleteMessageBatchResult> deleteFuture = new SqsFutureReceiver<>();
        sqsClient.deleteMessageBatchAsync(
                new DeleteMessageBatchRequest().withQueueUrl(queueURL).withEntries(deleteEntries), deleteFuture);
        return deleteFuture;
    }

    private List<SolrInputDocument> convertSqsMessageToSolrJob(final List<Message> messages) {
        logger.trace("Create Solr request");
        List<SolrInputDocument> solrJobs = messages.stream().flatMap(sqsMessage -> {
            // uncompress the messages
            String snsMsg = sqsMessage.getBody();
            List<Map<String, Object>> maps = MessageUtil.decodeMessages(snsMsg);

            return maps.stream().map(map -> {
                return messageService.convertHashMap2SolrInputDocument(map);
            });
        }).collect(Collectors.toList());
        return solrJobs;
    }

}

That is the codes.

References:
https://lucene.apache.org/solr/guide/7_1/solrcloud-recoveries-and-write-tolerance.html