Merge pull request #22 from vivetiwa/pause-connector-500-or-authentication-error

Pause Connector in case of 500 and unauthorised error

Author: nipunnair

gradle.properties

@@ -11,7 +11,7 @@
 ##
 
 group=com.adobe.platform.streaming
-versionMain=0.0.3
+versionMain=0.0.5
 versionQualifier=
 
 param_artifactory_user=

streaming-connect-common/src/main/java/com/adobe/platform/streaming/auth/AbstractAuthProvider.java

@@ -58,7 +58,7 @@ protected ContentHandler<TokenResponse> getContentHandler() {
       public TokenResponse getContent(HttpConnection conn) throws HttpException {
         try (InputStream in = conn.getInputStream()) {
           return mapper.readValue(in, TokenResponse.class);
-        } catch (HttpException | IOException e) {
+        } catch (IOException e) {
           throw new HttpException("Error parsing response", e);
         }
       }

streaming-connect-common/src/main/java/com/adobe/platform/streaming/auth/impl/IMSTokenProvider.java

@@ -50,7 +50,7 @@ public class IMSTokenProvider extends AbstractAuthProvider {
   @Override
   protected TokenResponse getTokenResponse() throws AuthException {
     LOG.debug("refreshing expired accessToken: {}", clientId);
-    StringBuffer params = new StringBuffer()
+    StringBuilder params = new StringBuilder()
       .append("grant_type=authorization_code")
       .append("&client_id=").append(clientId)
       .append("&client_secret=").append(clientSecret)

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/ContentHandler.java

@@ -35,7 +35,7 @@ public JSONObject getContent(HttpConnection conn) throws HttpException {
       try (InputStream in = conn.getInputStream()) {
         return new JSONObject(HttpUtil.streamToString(in));
       } catch (IOException e) {
-        throw new HttpException("Error parsing content", e);
+        throw new HttpException("Error parsing content", e, 405);
       }
     }
   };

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpConnection.java

@@ -64,6 +64,9 @@ private HttpConnection() {
   @SuppressWarnings("squid:S3776")
   HttpURLConnection connect() throws HttpException {
     Throwable cause = null;
+    int responseCode = 500;
+    String errorMsg = "";
+
     while (retries++ < maxRetries) {
       try {
         URL request = new URL(new URL(endpoint), url);
@@ -104,18 +107,21 @@ HttpURLConnection connect() throws HttpException {
           }
         }
 
-        int responseCode = conn.getResponseCode();
+        responseCode = conn.getResponseCode();
         if (HttpUtil.is2xx(responseCode)) {
           break;
         }
 
-        String errorMsg = errorStreamToString();
+        errorMsg = errorStreamToString();
         if (HttpUtil.is5xx(responseCode)) {
           LOG.warn("attempt {} of {} failed with {} response - {}", retries, maxRetries, responseCode, errorMsg);
           close();
           HttpUtil.sleepUninterrupted(retryBackoff);
+        } else if (HttpUtil.isUnauthorized(responseCode)) {
+          throw new AuthException(String.format("requested failed. unauthorized to access the endpoint. " +
+            "response code %s", responseCode));
         } else {
-          throw new HttpException("request failed (" + responseCode + "): " + errorMsg);
+          throw new HttpException("request failed (" + responseCode + "): " + errorMsg, responseCode);
         }
       } catch (MalformedURLException e) {
         throw new HttpException(("bad withUrl: " + url), e);
@@ -125,10 +131,14 @@ HttpURLConnection connect() throws HttpException {
         cause = e;
         HttpUtil.sleepUninterrupted(retryBackoff);
       } catch (AuthException authException) {
-        throw new HttpException("exception while fetching the auth token", authException);
+        throw new HttpException("exception while fetching the auth token", authException, responseCode);
       }
     }
 
+    if (HttpUtil.is5xx(responseCode)) {
+      throw new HttpException("request failed (" + responseCode + "): " + errorMsg, responseCode);
+    }
+
     if (conn == null) {
       throw new HttpException("unable to connect", cause);
     }

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpException.java

@@ -17,14 +17,27 @@
  */
 public class HttpException extends Exception {
 
-  public HttpException() {}
+  private int responseCode = 500;
 
   public HttpException(String message) {
     super(message);
   }
 
+  public HttpException(String message, int responseCode) {
+    super(message);
+    this.responseCode = responseCode;
+  }
+
   public HttpException(String message, Throwable cause) {
     super(message, cause);
   }
 
+  public HttpException(String message, Throwable cause, int responseCode) {
+    super(message, cause);
+    this.responseCode = responseCode;
+  }
+
+  public int getResponseCode() {
+    return responseCode;
+  }
 }

streaming-connect-common/src/main/java/com/adobe/platform/streaming/http/HttpUtil.java

@@ -21,16 +21,24 @@
 /**
  * @author Adobe Inc.
  */
-class HttpUtil {
+public class HttpUtil {
 
-  static boolean is2xx(int responseCode) {
+  public static boolean is2xx(int responseCode) {
     return (responseCode >= 200 && responseCode < 300);
   }
 
-  static boolean is5xx(int responseCode) {
+  public static boolean is5xx(int responseCode) {
     return (responseCode >= 500 && responseCode < 600);
   }
 
+  public static boolean is500(int responseCode) {
+    return responseCode == 500;
+  }
+
+  public static boolean isUnauthorized(int responseCode) {
+    return responseCode == 401 || responseCode == 403;
+  }
+
   static String streamToString(InputStream in) throws HttpException {
     StringBuilder sb = new StringBuilder(128);
     try (BufferedReader r = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/AbstractSinkTask.java

@@ -18,6 +18,7 @@
 import com.google.gson.GsonBuilder;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
 import org.slf4j.Logger;
@@ -115,17 +116,21 @@ public void put(Collection<SinkRecord> records) {
 
   public abstract int getPayloadLength(T dataToPublish);
 
-  public abstract void publishData(List<T> eventsToPublish);
+  public abstract void publishData(List<T> eventsToPublish) throws AEPStreamingException;
 
   private boolean flushNow(long tempCurrTime) {
     return tempCurrTime >= lastFlushMilliSec + flushIntervalMillis || bytesRead >= flushBytesCount;
   }
 
   private void publishAndLogIfRequired(List<T> eventsToPublish) {
+    try {
+      publishData(eventsToPublish);
+    } catch (AEPStreamingException e) {
+      throw new ConnectException("Failed to sink records.", e);
+    }
     if (LOG.isDebugEnabled()) {
       LOG.debug("ConnectorSinkTask: {} events sent to destination", eventsToPublish.size());
     }
-    publishData(eventsToPublish);
   }
 
   private void reset(List<T> eventsToPublish, long tempCurrentTime) {

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/DataPublisher.java

@@ -12,6 +12,8 @@
 
 package com.adobe.platform.streaming.sink;
 
+import com.adobe.platform.streaming.AEPStreamingException;
+
 import java.util.List;
 
 /**
@@ -21,7 +23,7 @@ public interface DataPublisher {
 
   void start();
 
-  void publishData(List<String> messages);
+  void publishData(List<String> messages) throws AEPStreamingException;
 
   void stop();
 }

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/impl/AEPPublisher.java

@@ -16,6 +16,7 @@
 import com.adobe.platform.streaming.http.ContentHandler;
 import com.adobe.platform.streaming.http.HttpException;
 import com.adobe.platform.streaming.http.HttpProducer;
+import com.adobe.platform.streaming.http.HttpUtil;
 import com.adobe.platform.streaming.sink.AbstractAEPPublisher;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -45,7 +46,7 @@ class AEPPublisher extends AbstractAEPPublisher {
   }
 
   @Override
-  public void publishData(List<String> messages) {
+  public void publishData(List<String> messages) throws AEPStreamingException {
     if (CollectionUtils.isEmpty(messages)) {
       LOG.debug("No messages to publish");
       return;
@@ -68,6 +69,9 @@ public void publishData(List<String> messages) {
       LOG.debug("Successfully published data to Adobe Experience Platform: {}", response);
     } catch (HttpException httpException) {
       LOG.error("Failed to publish data to Adobe Experience Platform", httpException);
+      if (HttpUtil.is500(httpException.getResponseCode()) || HttpUtil.isUnauthorized(httpException.getResponseCode())) {
+        throw new AEPStreamingException("Failed to publish", httpException);
+      }
     }
   }