From 4d25e875895ec82a50b2b5543ba2df8ef7bfc31e Mon Sep 17 00:00:00 2001
From: Mathias Fussenegger <f.mathias@zignar.net>
Date: Wed, 6 May 2015 21:21:33 +0200
Subject: [PATCH] handle EsRejectedExecutionException on blob upload

---
 .../main/java/io/crate/blob/BlobTransferTarget.java  |  4 ++--
 .../java/io/crate/http/netty/HttpBlobHandler.java    | 12 +++++++++++-
 2 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/blob/src/main/java/io/crate/blob/BlobTransferTarget.java b/blob/src/main/java/io/crate/blob/BlobTransferTarget.java
index 42090e13045d..5fafe486eaad 100644
--- a/blob/src/main/java/io/crate/blob/BlobTransferTarget.java
+++ b/blob/src/main/java/io/crate/blob/BlobTransferTarget.java
@@ -232,7 +232,7 @@ private void removeTransferAfterRecovery(UUID transferId) {
             }
         }
         if (toSchedule) {
-            logger.info("finished transfer {}, removing state", transferId);
+            logger.debug("finished transfer {}, removing state", transferId);
 
             /**
              * there might be a race condition that the recoveryActive flag is still false although a
@@ -340,7 +340,7 @@ public void stopRecovery() {
         synchronized (lock) {
             recoveryActive = false;
             for (UUID finishedUpload : finishedUploads) {
-                logger.info("finished transfer and recovery for {}, removing state", finishedUpload);
+                logger.debug("finished transfer and recovery for {}, removing state", finishedUpload);
                 activeTransfers.remove(finishedUpload);
             }
         }
diff --git a/blob/src/main/java/io/crate/http/netty/HttpBlobHandler.java b/blob/src/main/java/io/crate/http/netty/HttpBlobHandler.java
index 16d6ca2c8904..73d1070172ec 100644
--- a/blob/src/main/java/io/crate/http/netty/HttpBlobHandler.java
+++ b/blob/src/main/java/io/crate/http/netty/HttpBlobHandler.java
@@ -32,6 +32,7 @@
 import io.crate.blob.v2.BlobsDisabledException;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.indices.IndexMissingException;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
@@ -52,6 +53,7 @@
 import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.*;
 import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK;
 import static org.jboss.netty.handler.codec.http.HttpResponseStatus.PARTIAL_CONTENT;
+import static org.jboss.netty.handler.codec.http.HttpResponseStatus.TOO_MANY_REQUESTS;
 import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;
 
 public class HttpBlobHandler extends SimpleChannelUpstreamHandler implements
@@ -215,7 +217,12 @@ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
             logger.trace("channel closed: {}", ex.toString());
             return;
         } else if (ex instanceof IOException) {
-            logger.warn(ex.getMessage());
+            String message = ex.getMessage();
+            if (message != null && message.contains("Connection reset by peer")) {
+                logger.debug(message);
+            } else {
+                logger.warn(message, e);
+            }
             return;
         }
 
@@ -229,6 +236,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
         } else if (ex instanceof BlobsDisabledException || ex instanceof IndexMissingException) {
             status = HttpResponseStatus.BAD_REQUEST;
             body = ex.getMessage();
+        } else if (ex instanceof EsRejectedExecutionException) {
+            status = TOO_MANY_REQUESTS;
+            body = ex.getMessage();
         } else {
             status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
             logger.error("unhandled exception:", ex);