// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/aws/AmazonSNSNotification.java
// @Immutable
// @JsonIgnoreProperties(ignoreUnknown = true)
// @JsonInclude(JsonInclude.Include.NON_NULL)
// public final class AmazonSNSNotification {
// 
//   private final String message;
//   private final ZonedDateTime timestamp;
// 
//   /**
//    * Constructor
//    *
//    * @param message SNS message
//    * @param timestamp SNS timestamp
//    */
//   @JsonCreator
//   public AmazonSNSNotification(
//       @JsonProperty("Message") final String message,
//       @JsonProperty("Timestamp") final ZonedDateTime timestamp) {
//     this.message = message;
//     this.timestamp = timestamp;
//   }
// 
//   @JsonProperty("Message")
//   public String getMessage() {
//     return message;
//   }
// 
//   @JsonProperty("Timestamp")
//   public ZonedDateTime getTimestamp() {
//     return timestamp;
//   }
// 
//   @Nullable
//   @JsonIgnore
//   public Duration getDelayDuration() {
//     if (timestamp == null) {
//       return null;
//     }
//     return Duration.between(timestamp, ZonedDateTime.now(Clock.systemUTC()));
//   }
// 
//   @JsonIgnore
//   public boolean isValid() {
//     return message != null && timestamp != null;
//   }
// 
//   @Override
//   public boolean equals(final Object obj) {
//     if (this == obj) {
//       return true;
//     }
//     if ((obj == null) || (getClass() != obj.getClass())) {
//       return false;
//     }
// 
//     final AmazonSNSNotification other = (AmazonSNSNotification) obj;
//     return Objects.equals(message, other.message) && Objects.equals(timestamp, other.timestamp);
//   }
// 
//   @Override
//   public int hashCode() {
//     return Objects.hash(message, timestamp);
//   }
// 
//   @Override
//   public String toString() {
//     return MoreObjects.toStringHelper(this)
//         .add("message", message)
//         .add("timestamp", timestamp)
//         .toString();
//   }
// }
// 
// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/exceptions/AmazonS3ConstraintException.java
// public class AmazonS3ConstraintException extends Exception {
// 
//   private static final long serialVersionUID = 1L;
// 
//   public AmazonS3ConstraintException() {
//     super();
//   }
// 
//   public AmazonS3ConstraintException(final Throwable cause) {
//     super(cause);
//   }
// 
//   public AmazonS3ConstraintException(final String message) {
//     super(message);
//   }
// 
//   public AmazonS3ConstraintException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// }
// 
// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/exceptions/AmazonS3ZeroSizeException.java
// public class AmazonS3ZeroSizeException extends Exception {
// 
//   private static final long serialVersionUID = 1L;
// 
//   public AmazonS3ZeroSizeException() {
//     super();
//   }
// 
//   public AmazonS3ZeroSizeException(final Throwable cause) {
//     super(cause);
//   }
// 
//   public AmazonS3ZeroSizeException(final String message) {
//     super(message);
//   }
// 
//   public AmazonS3ZeroSizeException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// }


import static com.codahale.metrics.MetricRegistry.name;
import com.amazonaws.AmazonClientException;
import com.amazonaws.services.s3.event.S3EventNotification;
import com.amazonaws.services.s3.event.S3EventNotification.S3EventNotificationRecord;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectInputStream;
import com.amazonaws.services.sqs.model.Message;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.SharedMetricRegistries;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.smoketurner.pipeline.application.aws.AmazonSNSNotification;
import com.smoketurner.pipeline.application.exceptions.AmazonS3ConstraintException;
import com.smoketurner.pipeline.application.exceptions.AmazonS3ZeroSizeException;
import io.dropwizard.jackson.Jackson;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
    }

    // if we've processed all of the records, which includes skipping over
    // empty S3 files, the message has been fully processed.
    if (recordsProcessed == recordCount) {
      LOGGER.debug(
          "Processed {} of {} records, deleting SQS message", recordsProcessed, recordCount);
      return true;
    }

    LOGGER.debug(
        "Processed {} of {} records, not deleting SQS message: {}",
        recordsProcessed,
        recordCount,
        message.getMessageId());
    return false;
  }

  /**
   * Process an S3 event notification record by streaming object in {@link streamObject}
   *
   * @param record S3 event notification record
   * @return true if the record was fully processed, otherwise false
   */
  private boolean processRecord(@Nonnull final S3EventNotificationRecord record) {
    LOGGER.trace("Event Record: {}", record);

    final S3Object download;
    try {
      download = s3.fetch(record);
// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/aws/AmazonSNSNotification.java
// @Immutable
// @JsonIgnoreProperties(ignoreUnknown = true)
// @JsonInclude(JsonInclude.Include.NON_NULL)
// public final class AmazonSNSNotification {
// 
//   private final String message;
//   private final ZonedDateTime timestamp;
// 
//   /**
//    * Constructor
//    *
//    * @param message SNS message
//    * @param timestamp SNS timestamp
//    */
//   @JsonCreator
//   public AmazonSNSNotification(
//       @JsonProperty("Message") final String message,
//       @JsonProperty("Timestamp") final ZonedDateTime timestamp) {
//     this.message = message;
//     this.timestamp = timestamp;
//   }
// 
//   @JsonProperty("Message")
//   public String getMessage() {
//     return message;
//   }
// 
//   @JsonProperty("Timestamp")
//   public ZonedDateTime getTimestamp() {
//     return timestamp;
//   }
// 
//   @Nullable
//   @JsonIgnore
//   public Duration getDelayDuration() {
//     if (timestamp == null) {
//       return null;
//     }
//     return Duration.between(timestamp, ZonedDateTime.now(Clock.systemUTC()));
//   }
// 
//   @JsonIgnore
//   public boolean isValid() {
//     return message != null && timestamp != null;
//   }
// 
//   @Override
//   public boolean equals(final Object obj) {
//     if (this == obj) {
//       return true;
//     }
//     if ((obj == null) || (getClass() != obj.getClass())) {
//       return false;
//     }
// 
//     final AmazonSNSNotification other = (AmazonSNSNotification) obj;
//     return Objects.equals(message, other.message) && Objects.equals(timestamp, other.timestamp);
//   }
// 
//   @Override
//   public int hashCode() {
//     return Objects.hash(message, timestamp);
//   }
// 
//   @Override
//   public String toString() {
//     return MoreObjects.toStringHelper(this)
//         .add("message", message)
//         .add("timestamp", timestamp)
//         .toString();
//   }
// }
// 
// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/exceptions/AmazonS3ConstraintException.java
// public class AmazonS3ConstraintException extends Exception {
// 
//   private static final long serialVersionUID = 1L;
// 
//   public AmazonS3ConstraintException() {
//     super();
//   }
// 
//   public AmazonS3ConstraintException(final Throwable cause) {
//     super(cause);
//   }
// 
//   public AmazonS3ConstraintException(final String message) {
//     super(message);
//   }
// 
//   public AmazonS3ConstraintException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// }
// 
// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/exceptions/AmazonS3ZeroSizeException.java
// public class AmazonS3ZeroSizeException extends Exception {
// 
//   private static final long serialVersionUID = 1L;
// 
//   public AmazonS3ZeroSizeException() {
//     super();
//   }
// 
//   public AmazonS3ZeroSizeException(final Throwable cause) {
//     super(cause);
//   }
// 
//   public AmazonS3ZeroSizeException(final String message) {
//     super(message);
//   }
// 
//   public AmazonS3ZeroSizeException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// }


// Path: pipeline-application/src/main/java/com/smoketurner/pipeline/application/core/MessageProcessor.java
import static com.codahale.metrics.MetricRegistry.name;
import com.amazonaws.AmazonClientException;
import com.amazonaws.services.s3.event.S3EventNotification;
import com.amazonaws.services.s3.event.S3EventNotification.S3EventNotificationRecord;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectInputStream;
import com.amazonaws.services.sqs.model.Message;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.SharedMetricRegistries;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.smoketurner.pipeline.application.aws.AmazonSNSNotification;
import com.smoketurner.pipeline.application.exceptions.AmazonS3ConstraintException;
import com.smoketurner.pipeline.application.exceptions.AmazonS3ZeroSizeException;
import io.dropwizard.jackson.Jackson;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

    }

    // if we've processed all of the records, which includes skipping over
    // empty S3 files, the message has been fully processed.
    if (recordsProcessed == recordCount) {
      LOGGER.debug(
          "Processed {} of {} records, deleting SQS message", recordsProcessed, recordCount);
      return true;
    }

    LOGGER.debug(
        "Processed {} of {} records, not deleting SQS message: {}",
        recordsProcessed,
        recordCount,
        message.getMessageId());
    return false;
  }

  /**
   * Process an S3 event notification record by streaming object in {@link streamObject}
   *
   * @param record S3 event notification record
   * @return true if the record was fully processed, otherwise false
   */
  private boolean processRecord(@Nonnull final S3EventNotificationRecord record) {
    LOGGER.trace("Event Record: {}", record);

    final S3Object download;
    try {
      download = s3.fetch(record);
    } catch (AmazonS3ConstraintException | AmazonS3ZeroSizeException e) {