diff --git a/component/src/main/java/io/siddhi/extension/io/http/sink/HttpCallSink.java b/component/src/main/java/io/siddhi/extension/io/http/sink/HttpCallSink.java
index 9c3865ac..d2d234e4 100644
--- a/component/src/main/java/io/siddhi/extension/io/http/sink/HttpCallSink.java
+++ b/component/src/main/java/io/siddhi/extension/io/http/sink/HttpCallSink.java
@@ -522,7 +522,8 @@ public void publish(Object payload, DynamicOptions dynamicOptions, State state)
         }
     }
 
-    private void sendOauthRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList) {
+    private void sendOauthRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList)
+            throws ConnectionUnavailableException {
         //generate encoded base64 auth for getting refresh token
         String consumerKeyValue = consumerKey + ":" + consumerSecret;
         String encodedAuth = "Basic " + encodeBase64(consumerKeyValue);
@@ -543,13 +544,13 @@ private void sendOauthRequest(Object payload, DynamicOptions dynamicOptions, Lis
         } else {
             log.error("Error at sending oauth request to API endpoint: " +
                     publisherURL + "', with response code: " + response + ". Message dropped. Message dropped.");
-            throw new HttpSinkAdaptorRuntimeException("Error at sending oauth request to API endpoint: " +
+            throw new ConnectionUnavailableException("Error at sending oauth request to API endpoint: " +
                     publisherURL + "', with response code: " + response + ". Message dropped. Message dropped.");
         }
     }
 
     private void handleOAuthFailure(Object payload, DynamicOptions dynamicOptions, List<Header> headersList,
-                                    String encodedAuth) {
+                                    String encodedAuth) throws ConnectionUnavailableException {
 
         Boolean checkFromCache = accessTokenCache.checkAvailableKey(encodedAuth);
         if (checkFromCache) {
@@ -560,7 +561,7 @@ private void handleOAuthFailure(Object payload, DynamicOptions dynamicOptions, L
     }
 
     private void getNewAccessTokenWithCache(Object payload, DynamicOptions dynamicOptions, List<Header> headersList,
-                                            String encodedAuth) {
+                                            String encodedAuth) throws ConnectionUnavailableException {
         String accessToken = accessTokenCache.getAccessToken(encodedAuth);
         for (Header header : headersList) {
             if (header.getName().equals(HttpConstants.AUTHORIZATION_HEADER)) {
@@ -582,13 +583,13 @@ private void getNewAccessTokenWithCache(Object payload, DynamicOptions dynamicOp
         } else {
             log.error("Error at sending oauth request to API endpoint " + publisherURL + "', with response code: " +
                     response + ". Message dropped. ");
-            throw new HttpSinkAdaptorRuntimeException("Error at sending oauth request to API endpoint " +
+            throw new ConnectionUnavailableException("Error at sending oauth request to API endpoint " +
                     publisherURL + "', with response code: " + response + ". Message dropped.");
         }
     }
 
     private void requestForNewAccessToken(Object payload, DynamicOptions dynamicOptions, List<Header> headersList,
-                                          String encodedAuth) {
+                                          String encodedAuth) throws ConnectionUnavailableException {
         Boolean checkRefreshToken = accessTokenCache.checkRefreshAvailableKey(encodedAuth);
         if (checkRefreshToken) {
             for (Header header : headersList) {
@@ -633,7 +634,7 @@ private void requestForNewAccessToken(Object payload, DynamicOptions dynamicOpti
             } else {
                 log.error("Error at sending oauth request to API endpoint " + publisherURL + "', with response code: " +
                         response + ". Message dropped.");
-                throw new HttpSinkAdaptorRuntimeException("Error at sending oauth request to API endpoint  " +
+                throw new ConnectionUnavailableException("Error at sending oauth request to API endpoint  " +
                         publisherURL + "', with response code: " + response + ". Message dropped.");
             }
 
@@ -648,13 +649,14 @@ private void requestForNewAccessToken(Object payload, DynamicOptions dynamicOpti
         } else {
             log.error("Failed to generate new access token for the expired access token. Error code: " +
                     accessTokenCache.getResponseCode(encodedAuth) + ". Message dropped.");
-            throw new HttpSinkAdaptorRuntimeException("Failed to generate new access token for the expired" +
+            throw new ConnectionUnavailableException("Failed to generate new access token for the expired" +
                     " access token. Error code: " + accessTokenCache.getResponseCode(encodedAuth) +
                     ". Message dropped.");
         }
     }
 
-    private int sendRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList, int tryCount) {
+    private int sendRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList, int tryCount)
+            throws ConnectionUnavailableException {
         if (!publisherURLOption.isStatic()) {
             super.initClientConnector(dynamicOptions);
         }
@@ -688,13 +690,13 @@ private int sendRequest(Object payload, DynamicOptions dynamicOptions, List<Head
                 boolean latchCount = latch.await(30, TimeUnit.SECONDS);
                 if (!latchCount) {
                     log.debug("Time out due to getting getting response from " + publisherURL + ". Message dropped.");
-                    throw new HttpSinkAdaptorRuntimeException("Time out due to getting getting response from "
+                    throw new ConnectionUnavailableException("Time out due to getting getting response from "
                             + publisherURL + ". Message dropped.");
 
                 }
             } catch (InterruptedException e) {
                 log.debug("Failed to get a response from " + publisherURL + "," + e + ". Message dropped.");
-                throw new HttpSinkAdaptorRuntimeException("Failed to get a response from " +
+                throw new ConnectionUnavailableException("Failed to get a response from " +
                         publisherURL + ", " + e + ". Message dropped.");
             }
             if (isBlockingIO) {
diff --git a/component/src/main/java/io/siddhi/extension/io/http/sink/HttpSink.java b/component/src/main/java/io/siddhi/extension/io/http/sink/HttpSink.java
index 2a053038..c9869aef 100644
--- a/component/src/main/java/io/siddhi/extension/io/http/sink/HttpSink.java
+++ b/component/src/main/java/io/siddhi/extension/io/http/sink/HttpSink.java
@@ -168,13 +168,6 @@
                         type = {DataType.STRING},
                         optional = true,
                         defaultValue = "-"),
-                @Parameter(
-                        name = "blocking.io",
-                        description = "Blocks the request thread until a response it received from HTTP " +
-                                "endpoint. This should be enabled for reliable messaging (error handling)",
-                        type = {DataType.BOOL},
-                        optional = true,
-                        defaultValue = "false"),
                 @Parameter(
                         name = "headers",
                         description = "HTTP request headers in format `\"'<key>:<value>','<key>:<value>'\"`.\n" +
@@ -475,7 +468,6 @@ public class HttpSink extends Sink {
     private long maxWaitTime;
     private String hostnameVerificationEnabled;
     private String sslVerificationDisabled;
-    private boolean isBlockingIO;
 
     private DefaultHttpWsConnectorFactory httpConnectorFactory;
 
@@ -595,9 +587,6 @@ protected StateFactory init(StreamDefinition outputStreamDefinition, OptionHolde
             authType = HttpConstants.NO_AUTH;
         }
 
-        isBlockingIO = Boolean.parseBoolean(
-                optionHolder.validateAndGetStaticValue(HttpConstants.BLOCKING_IO, HttpConstants.FALSE));
-
         initConnectorFactory();
         if (publisherURLOption.isStatic()) {
             initClientConnector(null);
@@ -632,7 +621,8 @@ public void publish(Object payload, DynamicOptions dynamicOptions, State state)
         }
     }
 
-    private void sendOauthRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList) {
+    private void sendOauthRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList)
+            throws ConnectionUnavailableException {
         //generate encoded base64 auth for getting refresh token
         String consumerKeyValue = consumerKey + ":" + consumerSecret;
         String encodedAuth = "Basic " + encodeBase64(consumerKeyValue)
@@ -654,13 +644,13 @@ private void sendOauthRequest(Object payload, DynamicOptions dynamicOptions, Lis
         } else {
             log.error("Error at sending oauth request to API endpoint " +
                     publisherURL + "', with response code: " + response + ". Message dropped.");
-            throw new HttpSinkAdaptorRuntimeException("Error at sending oauth request to API endpoint " +
+            throw new ConnectionUnavailableException("Error at sending oauth request to API endpoint " +
                     publisherURL + "', and response code: " + response + ". Message dropped.");
         }
     }
 
     private void handleOAuthFailure(Object payload, DynamicOptions dynamicOptions, List<Header> headersList,
-                                    String encodedAuth) {
+                                    String encodedAuth) throws ConnectionUnavailableException {
         boolean checkFromCache = accessTokenCache.checkAvailableKey(encodedAuth);
         if (checkFromCache) {
             getNewAccessTokenWithCache(payload, dynamicOptions, headersList, encodedAuth);
@@ -670,7 +660,7 @@ private void handleOAuthFailure(Object payload, DynamicOptions dynamicOptions, L
     }
 
     private void getNewAccessTokenWithCache(Object payload, DynamicOptions dynamicOptions, List<Header> headersList,
-                                            String encodedAuth) {
+                                            String encodedAuth) throws ConnectionUnavailableException {
         String accessToken = accessTokenCache.getAccessToken(encodedAuth);
         for (Header header : headersList) {
             if (header.getName().equals(HttpConstants.AUTHORIZATION_HEADER)) {
@@ -692,13 +682,13 @@ private void getNewAccessTokenWithCache(Object payload, DynamicOptions dynamicOp
         } else {
             log.error("Error at sending oauth request to API endpoint " + publisherURL + "', with response code: " +
                     response + ". Message dropped.");
-            throw new HttpSinkAdaptorRuntimeException("Error at sending oauth request to API endpoint " + publisherURL +
+            throw new ConnectionUnavailableException("Error at sending oauth request to API endpoint " + publisherURL +
                     "', with response code: " + response + ". Message dropped.");
         }
     }
 
     private void requestForNewAccessToken(Object payload, DynamicOptions dynamicOptions, List<Header> headersList,
-                                          String encodedAuth) {
+                                          String encodedAuth) throws ConnectionUnavailableException {
         Boolean checkRefreshToken = accessTokenCache.checkRefreshAvailableKey(encodedAuth);
         if (checkRefreshToken) {
             for (Header header : headersList) {
@@ -744,7 +734,7 @@ private void requestForNewAccessToken(Object payload, DynamicOptions dynamicOpti
             } else {
                 log.error("Error at sending oauth request to API endpoint " + publisherURL + "', with response code: " +
                         response + ". Message dropped.");
-                throw new HttpSinkAdaptorRuntimeException("Error at sending oauth request to API endpoint " +
+                throw new ConnectionUnavailableException("Error at sending oauth request to API endpoint " +
                         publisherURL + "', with response code: " + response + ". Message dropped.");
             }
         } else if (accessTokenCache.getResponseCode(encodedAuth) == HttpConstants.AUTHENTICATION_FAIL_CODE) {
@@ -758,7 +748,7 @@ private void requestForNewAccessToken(Object payload, DynamicOptions dynamicOpti
         } else {
             log.error("Failed to generate new access token for the expired access token. Error code: " +
                     accessTokenCache.getResponseCode(encodedAuth) + ". Message dropped.");
-            throw new HttpSinkAdaptorRuntimeException("Failed to generate new access token for the expired" +
+            throw new ConnectionUnavailableException("Failed to generate new access token for the expired" +
                     " access token. Error code: " + accessTokenCache.getResponseCode(encodedAuth)
                     + ". Message dropped.");
         }
@@ -782,7 +772,7 @@ void getAccessToken(DynamicOptions dynamicOptions, String encodedAuth, String to
     }
 
     void setAccessToken(String encodedAuth, DynamicOptions dynamicOptions,
-                        List<Header> headersList) {
+                        List<Header> headersList) throws ConnectionUnavailableException {
         //check the availability of the authorization
         String accessToken;
         boolean authAvailability = false;
@@ -823,7 +813,7 @@ void setAccessToken(String encodedAuth, DynamicOptions dynamicOptions,
             } else {
                 log.error("Failed to generate new access token for the expired access token. Error code: " +
                         accessTokenCache.getResponseCode(encodedAuth) + ". Message dropped.");
-                throw new HttpSinkAdaptorRuntimeException("Failed to generate new access token for the expired" +
+                throw new ConnectionUnavailableException("Failed to generate new access token for the expired" +
                         " access token. Error code: " + accessTokenCache.getResponseCode(encodedAuth) +
                         ". Message dropped.");
             }
@@ -841,7 +831,8 @@ void setAccessToken(String encodedAuth, DynamicOptions dynamicOptions,
         }
     }
 
-    private int sendRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList) {
+    private int sendRequest(Object payload, DynamicOptions dynamicOptions, List<Header> headersList)
+            throws ConnectionUnavailableException {
         if (!publisherURLOption.isStatic()) {
             initClientConnector(dynamicOptions);
         }
@@ -872,45 +863,21 @@ private int sendRequest(Object payload, DynamicOptions dynamicOptions, List<Head
                 boolean latchCount = latch.await(30, TimeUnit.SECONDS);
                 if (!latchCount) {
                     log.debug("Time out due to getting getting response from " + publisherURL + ". Message dropped.");
-                    throw new HttpSinkAdaptorRuntimeException("Time out due to getting getting response from "
+                    throw new ConnectionUnavailableException("Time out due to getting getting response from "
                             + publisherURL + ". Message dropped.");
                 }
             } catch (InterruptedException e) {
                 log.debug("Failed to get a response from " + publisherURL + "," + e + ". Message dropped.");
-                throw new HttpSinkAdaptorRuntimeException("Failed to get a response from " +
+                throw new ConnectionUnavailableException("Failed to get a response from " +
                         publisherURL + ", " + e + ". Message dropped.");
             }
             HttpCarbonMessage response = listener.getHttpResponseMessage();
             return response.getNettyHttpResponse().status().code();
-        } else if (!isBlockingIO) {
-            clientConnector.send(cMessage);
-            return HttpConstants.SUCCESS_CODE;
         } else {
-            CountDownLatch latch = new CountDownLatch(1);
             HttpResponseFuture responseFuture = clientConnector.send(cMessage);
-            HTTPResponseListener responseListener = new HTTPResponseListener(latch);
+            HTTPResponseListener responseListener = new HTTPResponseListener(payload, dynamicOptions, this);
             responseFuture.setHttpConnectorListener(responseListener);
-
-            try {
-                boolean latchCount = latch.await(30, TimeUnit.SECONDS);
-                if (!latchCount) {
-                    log.debug("Time out due to getting getting response from " + publisherURL + ". Message dropped.");
-                    throw new HttpSinkAdaptorRuntimeException("Time out due to getting getting response from "
-                            + publisherURL + ". Message dropped.");
-
-                }
-            } catch (InterruptedException e) {
-                log.debug("Failed to get a response from " + publisherURL + "," + e + ". Message dropped.");
-                throw new HttpSinkAdaptorRuntimeException("Failed to get a response from " +
-                        publisherURL + ", " + e + ". Message dropped.");
-            }
-
-            if (responseListener.throwable == null) {
-                return HttpConstants.SUCCESS_CODE;
-            } else {
-                throw new SiddhiAppRuntimeException("Siddhi App " + siddhiAppContext.getName() + " failed to publish " +
-                        "events to HTTP endpoint", responseListener.throwable);
-            }
+            return HttpConstants.SUCCESS_CODE;
         }
     }
 
@@ -1154,22 +1121,26 @@ private String encodeBase64(String consumerKeyValue) {
     }
 
     static class HTTPResponseListener implements HttpConnectorListener {
-        Throwable throwable;
-        CountDownLatch countDownLatch;
+        Object payload;
+        DynamicOptions dynamicOptions;
+        HttpSink httpSink;
 
-        HTTPResponseListener(CountDownLatch latch) {
-            this.countDownLatch = latch;
+        HTTPResponseListener(Object payload, DynamicOptions dynamicOptions, HttpSink httpSink) {
+            this.payload = payload;
+            this.dynamicOptions = dynamicOptions;
+            this.httpSink = httpSink;
         }
 
         @Override
         public void onMessage(HttpCarbonMessage httpCarbonMessage) {
-            countDownLatch.countDown();
+
         }
 
         @Override
         public void onError(Throwable throwable) {
-            this.throwable = throwable;
-            countDownLatch.countDown();
+            httpSink.onError(payload, dynamicOptions,
+                    new ConnectionUnavailableException("Siddhi App " + httpSink.siddhiAppContext.getName() +
+                            " failed to publish events to HTTP endpoint: " + httpSink.publisherURL, throwable));
         }
     }
 }
diff --git a/component/src/test/java/io/siddhi/extension/io/http/sink/HttpSinkTestCase.java b/component/src/test/java/io/siddhi/extension/io/http/sink/HttpSinkTestCase.java
index ca79c6bc..839b398a 100644
--- a/component/src/test/java/io/siddhi/extension/io/http/sink/HttpSinkTestCase.java
+++ b/component/src/test/java/io/siddhi/extension/io/http/sink/HttpSinkTestCase.java
@@ -21,8 +21,8 @@
 import com.sun.net.httpserver.Headers;
 import io.siddhi.core.SiddhiAppRuntime;
 import io.siddhi.core.SiddhiManager;
-import io.siddhi.core.stream.StreamJunction;
 import io.siddhi.core.stream.input.InputHandler;
+import io.siddhi.core.stream.output.sink.Sink;
 import io.siddhi.extension.io.http.sink.util.HttpServerListenerHandler;
 import io.siddhi.extension.io.http.sink.util.UnitTestAppender;
 import io.siddhi.extension.map.xml.sinkmapper.XMLSinkMapper;
@@ -168,7 +168,7 @@ public void testHTTPConnectionFailure() throws InterruptedException {
         SiddhiManager siddhiManager = new SiddhiManager();
         siddhiManager.setExtension("xml-output-mapper", XMLSinkMapper.class);
         String inStreamDefinition = "Define stream FooStream (message String,method String,headers String);"
-                + "@sink(type='http',blocking.io='true',publisher.url='http://localhost:8010/abcd',method='{{method}}',"
+                + "@sink(type='http', publisher.url='http://localhost:8010/abcd',method='{{method}}',"
                 + "headers='{{headers}}',"
                 + "@map(type='xml', @payload('{{message}}'))) "
                 + "Define stream BarStream (message String,method String,headers String);";
@@ -182,16 +182,16 @@ public void testHTTPConnectionFailure() throws InterruptedException {
         SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(inStreamDefinition + query);
         InputHandler fooStream = siddhiAppRuntime.getInputHandler("FooStream");
         siddhiAppRuntime.start();
-        Logger logger = Logger.getLogger(StreamJunction.class);
+        Logger logger = Logger.getLogger(Sink.class);
         UnitTestAppender appender = new UnitTestAppender();
         logger.addAppender(appender);
 
         try {
             fooStream.send(1566562744069L, new Object[]{payload, "POST", "'Name:John','Age:23'"});
-            String expectedMessage = "failed to publish events to HTTP endpoint. Hence, dropping event " +
-                    "'StreamEvent{ timestamp=1566562744069, beforeWindowData=null, onAfterWindowData=null, " +
-                    "outputData=[<events><event><symbol>WSO2</symbol><price>55.645</price><volume>100</volume>" +
-                    "</event></events>, POST, 'Name:John','Age:23'], type=CURRENT, next=null}";
+            Thread.sleep(3000);
+            String expectedMessage = "Dropping event at Sink 'http' at 'BarStream' as its still trying to " +
+                    "reconnect!, events dropped '<events><event><symbol>WSO2</symbol><price>55.645</price>" +
+                    "<volume>100</volume></event></events>'";
             Assert.assertTrue(appender.getMessages().contains(expectedMessage));
         } finally {
             logger.removeAppender(appender);
diff --git a/pom.xml b/pom.xml
index e6e273aa..a21e7c2c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -415,7 +415,7 @@
         <siddhi.mapper.json.version>5.0.2</siddhi.mapper.json.version>
         <siddhi.mapper.text.version>2.0.1</siddhi.mapper.text.version>
         <siddhi.mapper.xml.version>5.0.2</siddhi.mapper.xml.version>
-        <siddhi.version>5.1.2</siddhi.version>
+        <siddhi.version>5.1.4</siddhi.version>
         <siddhi.version.range>[5.0.0,6.0.0)</siddhi.version.range>
         <siddhi.distribution.version>1.0.0-m3</siddhi.distribution.version>
         <pax.exam.spi.version>4.10.0</pax.exam.spi.version>