azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/utils/AbstractMailer.java b/azkaban-common/src/main/java/azkaban/utils/AbstractMailer.java
index a9074a7..681ae33 100644
--- a/azkaban-common/src/main/java/azkaban/utils/AbstractMailer.java
+++ b/azkaban-common/src/main/java/azkaban/utils/AbstractMailer.java
@@ -19,6 +19,7 @@ package azkaban.utils;
 import java.util.Collection;
 
 public class AbstractMailer {
+  private static int MB_IN_BYTES = 1048576;
   private String clientHostname;
   private int clientPort;
   private boolean usesSSL;
@@ -31,11 +32,18 @@ public class AbstractMailer {
 
   private String referenceURL;
 
+  private long attachmentMazSizeInByte;
+
   public AbstractMailer(Props props) {
     this.azkabanName = props.getString("azkaban.name", "azkaban");
     this.mailHost = props.getString("mail.host", "localhost");
     this.mailUser = props.getString("mail.user", "");
     this.mailPassword = props.getString("mail.password", "");
+    long maxAttachmentSizeInMB =
+        props.getInt("mail.max.attachment.size.mb", 1000);
+
+    attachmentMazSizeInByte = maxAttachmentSizeInMB * MB_IN_BYTES;
+
     this.mailSender = props.getString("mail.sender", "");
 
     this.clientHostname = props.get("server.hostname");
@@ -92,4 +100,13 @@ public class AbstractMailer {
   public String getMailSender() {
     return mailSender;
   }
+
+  /**
+   * Attachment maximum size in bytes
+   * 
+   * @return
+   */
+  public long getAttachmentMaxSize() {
+    return attachmentMazSizeInByte;
+  }
 }

diff --git a/azkaban-common/src/main/java/azkaban/utils/Emailer.java b/azkaban-common/src/main/java/azkaban/utils/Emailer.java
index 947800a..ee0dbad 100644
--- a/azkaban-common/src/main/java/azkaban/utils/Emailer.java
+++ b/azkaban-common/src/main/java/azkaban/utils/Emailer.java
@@ -16,7 +16,6 @@
 
 package azkaban.utils;
 
-import java.lang.String;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -32,9 +31,6 @@ import azkaban.executor.Status;
 import azkaban.executor.mail.DefaultMailCreator;
 import azkaban.executor.mail.MailCreator;
 import azkaban.sla.SlaOption;
-import azkaban.utils.AbstractMailer;
-import azkaban.utils.EmailMessage;
-import azkaban.utils.Props;
 
 public class Emailer extends AbstractMailer implements Alerter {
   private static Logger logger = Logger.getLogger(Emailer.class);
@@ -71,6 +67,8 @@ public class Emailer extends AbstractMailer implements Alerter {
         props.getInt("mail.connection.timeout.millis", 10000);
     EmailMessage.setConnectionTimeout(connectionTimeout);
 
+    EmailMessage.setAttachmentMaxSize(getAttachmentMaxSize());
+
     this.clientHostname = props.getString("jetty.hostname", "localhost");
 
     if (props.getBoolean("jetty.use.ssl", true)) {

diff --git a/azkaban-common/src/main/java/azkaban/utils/EmailMessage.java b/azkaban-common/src/main/java/azkaban/utils/EmailMessage.java
index c939b13..4826478 100644
--- a/azkaban-common/src/main/java/azkaban/utils/EmailMessage.java
+++ b/azkaban-common/src/main/java/azkaban/utils/EmailMessage.java
@@ -18,6 +18,7 @@ package azkaban.utils;
 
 import java.io.File;
 import java.io.InputStream;
+import java.net.SocketTimeoutException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Date;
@@ -36,9 +37,13 @@ import javax.mail.internet.MimeBodyPart;
 import javax.mail.internet.MimeMessage;
 import javax.mail.internet.MimeMultipart;
 
+import org.apache.log4j.Logger;
+
 import com.sun.mail.smtp.SMTPTransport;
 
 public class EmailMessage {
+  private final Logger logger = Logger.getLogger(EmailMessage.class);
+
   private static String protocol = "smtp";
   private List<String> _toAddress = new ArrayList<String>();
   private String _mailHost;
@@ -48,9 +53,11 @@ public class EmailMessage {
   private String _fromAddress;
   private String _mimeType = "text/plain";
   private String _tls;
+
   private StringBuffer _body = new StringBuffer();
   private static int _mailTimeout = 10000;
   private static int _connectionTimeout = 10000;
+  private static long _attachmentMaxSizeInByte = 134217728; // 1 GB
 
   private ArrayList<BodyPart> _attachments = new ArrayList<BodyPart>();
 
@@ -72,6 +79,14 @@ public class EmailMessage {
     _connectionTimeout = timeoutMillis;
   }
 
+  public static void setAttachmentMaxSize(long sizeInByte) {
+    if (sizeInByte < 1) {
+      throw new IllegalArgumentException(
+          "attachment max size can't be 0 or negative");
+    }
+    _attachmentMaxSizeInByte = sizeInByte;
+  }
+
   public EmailMessage setMailHost(String host) {
     _mailHost = host;
     return this;
@@ -118,6 +133,12 @@ public class EmailMessage {
 
   public EmailMessage addAttachment(String attachmentName, File file)
       throws MessagingException {
+
+    if (file.length() > _attachmentMaxSizeInByte) {
+      throw new MessageAttachmentExceededMaximumSizeException("Attachment '"
+          + attachmentName + "' exceeded maximum size "
+          + _attachmentMaxSizeInByte);
+    }
     BodyPart attachmentPart = new MimeBodyPart();
     DataSource fileDataSource = new FileDataSource(file);
     attachmentPart.setDataHandler(new DataHandler(fileDataSource));
@@ -194,7 +215,23 @@ public class EmailMessage {
     // Transport transport = session.getTransport();
 
     SMTPTransport t = (SMTPTransport) session.getTransport(protocol);
-    t.connect(_mailHost, _mailUser, _mailPassword);
+    try {
+      t.connect(_mailHost, _mailUser, _mailPassword);
+    } catch (MessagingException ste) {
+      if (ste.getCause() instanceof SocketTimeoutException) {
+        try {
+          // retry on SocketTimeoutException
+          t.connect(_mailHost, _mailUser, _mailPassword);
+          logger
+              .info("Email retry on SocketTimeoutException completed successfully");
+        } catch (MessagingException me) {
+          logger.info("Email retry on SocketTimeoutException failed");
+          throw me;
+        }
+      } else {
+        throw ste;
+      }
+    }
     t.sendMessage(message, message.getRecipients(Message.RecipientType.TO));
     t.close();
   }

diff --git a/azkaban-common/src/main/java/azkaban/utils/MessageAttachmentExceededMaximumSizeException.java b/azkaban-common/src/main/java/azkaban/utils/MessageAttachmentExceededMaximumSizeException.java
new file mode 100644
index 0000000..344bc86
--- /dev/null
+++ b/azkaban-common/src/main/java/azkaban/utils/MessageAttachmentExceededMaximumSizeException.java
@@ -0,0 +1,26 @@
+package azkaban.utils;
+
+import javax.mail.MessagingException;
+
+/**
+ * To indicate the attachment size is larger than allowed size
+ * 
+ * @author hluu
+ *
+ */
+public class MessageAttachmentExceededMaximumSizeException extends
+    MessagingException {
+
+  public MessageAttachmentExceededMaximumSizeException() {
+    super();
+  }
+
+  public MessageAttachmentExceededMaximumSizeException(String s) {
+    super(s);
+  }
+
+  public MessageAttachmentExceededMaximumSizeException(String s, Exception e) {
+    super(s, e);
+  }
+
+}