<commit_before>/*
 * Copyright 2017 StreamSets Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.streamsets.pipeline.stage.origin.s3;

import com.amazonaws.AmazonClientException;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.codahale.metrics.Meter;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.streamsets.pipeline.api.BatchContext;
import com.streamsets.pipeline.api.PushSource;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.config.PostProcessingOptions;
import com.streamsets.pipeline.lib.util.AntPathMatcher;
import com.streamsets.pipeline.stage.lib.aws.AWSUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class S3Spooler {

  private static final Logger LOG = LoggerFactory.getLogger(S3Spooler.class);

  private final PushSource.Context context;
  private final S3ConfigBean s3ConfigBean;
  private final AmazonS3 s3Client;
  private AntPathMatcher pathMatcher;
  private AtomicBoolean filling;
  private volatile S3Offset lastElementAddedToQueue;
  private volatile boolean newDataAfterEventSent;

  public S3Spooler(PushSource.Context context, S3ConfigBean s3ConfigBean) {
    this.context = context;
    this.s3ConfigBean = s3ConfigBean;
    this.s3Client = s3ConfigBean.s3Config.getS3Client();
    lastElementAddedToQueue = null;
    newDataAfterEventSent = true;
  }

  private volatile S3ObjectSummary currentObject;
  private ArrayBlockingQueue<S3ObjectSummary> objectQueue;
  private Meter spoolQueueMeter;

  public void init() {
    try {
      filling = new AtomicBoolean(false);
      objectQueue = new ArrayBlockingQueue<>(s3ConfigBean.s3FileConfig.poolSize);
      spoolQueueMeter = context.createMeter("spoolQueue");
      pathMatcher = new AntPathMatcher(s3ConfigBean.s3Config.delimiter);
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }

  public void destroy() {
    if(objectQueue != null) {
      objectQueue.clear();
      objectQueue = null;
    }
  }

  private void findAndQueueObjects(
      AmazonS3Source amazonS3Source, BatchContext batchContext
  ) throws AmazonClientException {
    S3Offset s3offset;
    if (lastElementAddedToQueue != null) {
      s3offset = lastElementAddedToQueue;
    } else {
      s3offset = amazonS3Source.getLatestOffset();
    }

    List<S3ObjectSummary> s3ObjectSummaries;
    if (!AWSUtil.containsWildcard(s3ConfigBean.s3FileConfig.prefixPattern)) {
      // No wildcard in the prefixPattern - don't need to scan the bucket
      s3ObjectSummaries = AmazonS3Util.getObjectNoWildcard(s3Client, s3ConfigBean, s3offset, s3ConfigBean.s3FileConfig.prefixPattern);
    } else {
      ObjectOrdering objectOrdering = s3ConfigBean.s3FileConfig.objectOrdering;
      switch (objectOrdering) {
        case TIMESTAMP:
          s3ObjectSummaries = AmazonS3Util.listObjectsChronologically(s3Client,
              s3ConfigBean,
              pathMatcher,
              s3offset,
              objectQueue.remainingCapacity()
          );
          break;
        case LEXICOGRAPHICAL:
          s3ObjectSummaries = AmazonS3Util.listObjectsLexicographically(s3Client,
              s3ConfigBean,
              pathMatcher,
              s3offset,
              objectQueue.remainingCapacity()
          );
          break;
        default:
          throw new IllegalArgumentException("Unknown ordering: " + objectOrdering.getLabel());
      }
    }
    for (S3ObjectSummary objectSummary : s3ObjectSummaries) {
      addObjectToQueue(objectSummary);
    }
    spoolQueueMeter.mark(objectQueue.size());
    LOG.debug("Found '{}' files", objectQueue.size());
    if (s3ObjectSummaries.isEmpty()) {
      // Before sending the event we will check that all the threads have finished with their objects, if yes, we
      // send the event as normal, if not we will skip and try to send it again if the queue is still empty when the<commit_msg>SDC-15066. Fix amazon S3 bug when processing a single file<commit_after>/*
 * Copyright 2017 StreamSets Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.streamsets.pipeline.stage.origin.s3;

import com.amazonaws.AmazonClientException;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.codahale.metrics.Meter;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.streamsets.pipeline.api.BatchContext;
import com.streamsets.pipeline.api.PushSource;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.config.PostProcessingOptions;
import com.streamsets.pipeline.lib.util.AntPathMatcher;
import com.streamsets.pipeline.stage.lib.aws.AWSUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class S3Spooler {

  private static final Logger LOG = LoggerFactory.getLogger(S3Spooler.class);

  private final PushSource.Context context;
  private final S3ConfigBean s3ConfigBean;
  private final AmazonS3 s3Client;
  private AntPathMatcher pathMatcher;
  private AtomicBoolean filling;
  private volatile S3Offset lastElementAddedToQueue;
  private volatile boolean newDataAfterEventSent;

  public S3Spooler(PushSource.Context context, S3ConfigBean s3ConfigBean) {
    this.context = context;
    this.s3ConfigBean = s3ConfigBean;
    this.s3Client = s3ConfigBean.s3Config.getS3Client();
    lastElementAddedToQueue = null;
    newDataAfterEventSent = true;
  }

  private volatile S3ObjectSummary currentObject;
  private ArrayBlockingQueue<S3ObjectSummary> objectQueue;
  private Meter spoolQueueMeter;

  public void init() {
    try {
      filling = new AtomicBoolean(false);
      objectQueue = new ArrayBlockingQueue<>(s3ConfigBean.s3FileConfig.poolSize);
      spoolQueueMeter = context.createMeter("spoolQueue");
      pathMatcher = new AntPathMatcher(s3ConfigBean.s3Config.delimiter);
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }

  public void destroy() {
    if(objectQueue != null) {
      objectQueue.clear();
      objectQueue = null;
    }
  }

  private void findAndQueueObjects(
      AmazonS3Source amazonS3Source, BatchContext batchContext
  ) throws AmazonClientException {
    S3Offset s3offset;
    if (lastElementAddedToQueue != null) {
      s3offset = lastElementAddedToQueue;
    } else {
      s3offset = amazonS3Source.getLatestOffset();
    }

    List<S3ObjectSummary> s3ObjectSummaries = Collections.emptyList();
    if (!AWSUtil.containsWildcard(s3ConfigBean.s3FileConfig.prefixPattern)) {
      // No wildcard in the prefixPattern - don't need to scan the bucket
      try {
        s3ObjectSummaries = AmazonS3Util.getObjectNoWildcard(s3Client, s3ConfigBean, s3offset,
                s3ConfigBean.s3Config.commonPrefix + s3ConfigBean.s3FileConfig.prefixPattern);
      } catch (AmazonS3Exception e) {
        LOG.warn("Object '{}' not found", s3ConfigBean.s3Config.commonPrefix + s3ConfigBean.s3FileConfig.prefixPattern);
      }

    } else {
      ObjectOrdering objectOrdering = s3ConfigBean.s3FileConfig.objectOrdering;
      switch (objectOrdering) {
        case TIMESTAMP:
          s3ObjectSummaries = AmazonS3Util.listObjectsChronologically(s3Client,
              s3ConfigBean,
              pathMatcher,
              s3offset,
              objectQueue.remainingCapacity()
          );
          break;
        case LEXICOGRAPHICAL:
          s3ObjectSummaries = AmazonS3Util.listObjectsLexicographically(s3Client,
              s3ConfigBean,
              pathMatcher,
              s3offset,
              objectQueue.remainingCapacity()
          );
          break;
        default:
          throw new IllegalArgumentException("Unknown ordering: " + objectOrdering.getLabel());
      }
    }
    for (S3ObjectSummary objectSummary : s3ObjectSummaries) {
      addObjectToQueue(objectSummary);
    }
    spoolQueueMeter.mark(objectQueue.size());
    LOG.debug("Found '{}' files", objectQueue.size());
    if (s3ObjectSummaries.isEmpty()) {
      // Before sending the event we will check that all the threads have finished with their objects, if yes, we
      // send the event as normal, if not we will skip and try to send it again if the queue is still empty when the
{'fraction_non_alphanumeric': 0.081, 'fraction_numerical': 0.0161, 'mean_word_length': 3.6799251286850727, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2564043', 'n_tokens_mistral': 2900, 'n_tokens_neox': 2871, 'n_words': 701}