class SNSS3Handler extends BaseHandler<SNSEvent> implements Handler<SNSEvent> {

  @Override
  public void handler(SNSEvent event, Context context) throws HandlerException {
    if (!initialized) {
      init(context);
      SNSS3HandlerConfig handlerConfig = (SNSS3HandlerConfig) this.config.getHandlerConfig();
      this.logTrigger = handlerConfig.getLogSnsTrigger();
    }

    this.source = this.sources.get(0);
    this.inputFiles = new ArrayList<String>(0);

    if (this.logTrigger) {
      logger.info("trigger: " + gson.toJson(event));
    }

    for (SNSRecord record : event.getRecords()) {
      /*
       * Parse SNS as a S3 notification
       */
      String json = record.getSNS().getMessage();
      S3EventNotification s3Event = S3EventNotification.parseJson(json);

      /*
       * Validate the S3 file matches the regex
       */
      List<S3EventNotificationRecord> toProcess =
          new ArrayList<S3EventNotificationRecord>(s3Event.getRecords());
      for (S3EventNotificationRecord s3Record : s3Event.getRecords()) {
        String s3Path = String.format("s3://%s/%s", s3Record.getS3().getBucket().getName(),
            s3Record.getS3().getObject().getKey());
        try {
          this.source = SourceUtils.getSource(s3Path, this.sources);
        } catch (SourceNotFoundException e) {
          logger.warn("skipping processing " + s3Path);
          toProcess.remove(s3Record);
        }
      }

      if (toProcess.size() == 0) {
        logger.warn("Nothing to process");
        return;
      }

      this.inputFiles.addAll(toProcess.stream().map(m -> {
        return m.getS3().getObject().getKey();
      }).collect(Collectors.toList()));

      this.recordIterator =
          new S3EventIterator(new LambdaContext(context), toProcess, s3ClientFactory);

      super.process(context);
    }
  }

}