New release 1.0 -> 1.1:

- Extractor inferface no longer throws BackingStoreException. - Extractor, Transformer and Loader can all throw UnrecoverableStreamException which aborts stream. - Speed optimizations for simple transformations and loaders bypassing introspective mapping functions. - Support for java.time and Optional objects on the stream. - S3FastLoader will abort the stream if it fails to flush its buffer to S3. - RedshiftBulkLoader will abort the stream if it fails to write to S3 or there is a problem executing the COPY SQL on Redshift. - All stages will actively report a count of 0 on the metrics object even if they do not process any records.
amazon-archives · Mar 11, 2019 · 9935a76 · 9935a76
1 parent c03b2df
commit 9935a76
Show file tree

Hide file tree

Showing 46 changed files with 947 additions and 388 deletions.
diff --git a/pom.xml b/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>com.amazonaws</groupId>
     <artifactId>pocket-etl</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
 
     <licenses>
         <license>

diff --git a/src/main/java/com/amazon/pocketEtl/EtlCombineStage.java b/src/main/java/com/amazon/pocketEtl/EtlCombineStage.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -15,18 +15,18 @@
 
 package com.amazon.pocketEtl;
 
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nonnull;
+
 import com.amazon.pocketEtl.core.consumer.EtlConsumer;
 import com.amazon.pocketEtl.core.producer.EtlProducer;
+
 import lombok.AccessLevel;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
-import javax.annotation.Nonnull;
-import java.util.Collection;
-import java.util.stream.Collectors;
-
 @Getter(AccessLevel.PACKAGE)
-@EqualsAndHashCode(callSuper = true)
 class EtlCombineStage extends EtlProducerStage {
     private final static String DEFAULT_COMBINE_STAGE_NAME = "EtlStream.Combine";
 

diff --git a/src/main/java/com/amazon/pocketEtl/EtlConsumerStage.java b/src/main/java/com/amazon/pocketEtl/EtlConsumerStage.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -17,7 +17,6 @@
 
 import com.amazon.pocketEtl.core.DefaultLoggingStrategy;
 import com.amazon.pocketEtl.core.consumer.EtlConsumer;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
 import javax.annotation.Nonnull;
@@ -29,7 +28,6 @@
  * @param <T> The type of the data being operated on by this stage in the stream.
  */
 @Getter
-@EqualsAndHashCode
 public abstract class EtlConsumerStage<T> {
     /**
      * Static constructor for an EtlConsumerStage that transforms data in the stream. Used as a component in an

diff --git a/src/main/java/com/amazon/pocketEtl/EtlExtractStage.java b/src/main/java/com/amazon/pocketEtl/EtlExtractStage.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -15,21 +15,21 @@
 
 package com.amazon.pocketEtl;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nonnull;
+
 import com.amazon.pocketEtl.core.consumer.EtlConsumer;
 import com.amazon.pocketEtl.core.executor.EtlExecutorFactory;
 import com.amazon.pocketEtl.core.producer.EtlProducer;
 import com.amazon.pocketEtl.core.producer.EtlProducerFactory;
+
 import lombok.AccessLevel;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
-import javax.annotation.Nonnull;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.stream.Collectors;
-
 @Getter(AccessLevel.PACKAGE)
-@EqualsAndHashCode(callSuper = true)
 class EtlExtractStage extends EtlProducerStage {
     private final static String DEFAULT_EXTRACT_STAGE_NAME = "EtlStream.Extract";
     private final static EtlExecutorFactory defaultExecutorFactory = new EtlExecutorFactory();

diff --git a/src/main/java/com/amazon/pocketEtl/EtlLoadStage.java b/src/main/java/com/amazon/pocketEtl/EtlLoadStage.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -15,21 +15,21 @@
 
 package com.amazon.pocketEtl;
 
+import static org.apache.logging.log4j.LogManager.getLogger;
+
+import java.util.function.Function;
+
+import javax.annotation.Nonnull;
+
 import com.amazon.pocketEtl.core.consumer.EtlConsumer;
 import com.amazon.pocketEtl.core.consumer.EtlConsumerFactory;
 import com.amazon.pocketEtl.core.executor.EtlExecutor;
 import com.amazon.pocketEtl.core.executor.EtlExecutorFactory;
+
 import lombok.AccessLevel;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
-import javax.annotation.Nonnull;
-import java.util.function.Function;
-
-import static org.apache.logging.log4j.LogManager.getLogger;
-
 @Getter(AccessLevel.PACKAGE)
-@EqualsAndHashCode(callSuper = true)
 class EtlLoadStage<T> extends EtlConsumerStage<T> {
     private final static String DEFAULT_LOAD_STAGE_NAME = "EtlStream.Load";
 

diff --git a/src/main/java/com/amazon/pocketEtl/EtlProducerStage.java b/src/main/java/com/amazon/pocketEtl/EtlProducerStage.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -15,24 +15,24 @@
 
 package com.amazon.pocketEtl;
 
+import java.util.Arrays;
+import java.util.Collection;
+
+import javax.annotation.Nonnull;
+
 import com.amazon.pocketEtl.core.consumer.EtlConsumer;
 import com.amazon.pocketEtl.core.producer.EtlProducer;
+
 import lombok.AccessLevel;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
-import javax.annotation.Nonnull;
-import java.util.Arrays;
-import java.util.Collection;
-
 /**
  * Abstract class that represents the producer stage in an EtlStream. An EtlStream consists of a producer stage chained
  * to one or more consumer stages.
  */
 @Getter(AccessLevel.PACKAGE)
 @RequiredArgsConstructor
-@EqualsAndHashCode
 public abstract class EtlProducerStage {
     /**
      * Static constructor for an EtlProducerStage that extracts data and puts it on the stream. Used as a component in an

diff --git a/src/main/java/com/amazon/pocketEtl/EtlProfilingScope.java b/src/main/java/com/amazon/pocketEtl/EtlProfilingScope.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -46,10 +46,8 @@
  */
 public class EtlProfilingScope implements AutoCloseable {
     private final static Log logger = LogFactory.getLog(EtlProfilingScope.class);
-    private final static ThreadLocal<EtlMetrics> rootMetrics = new ThreadLocal<>();
     private final EtlMetrics ourMetrics;
     private final long metricsStart;
-    private final boolean isRoot;
     private final String scopeName;
     private boolean closed = false;
 
@@ -70,13 +68,6 @@ private EtlProfilingScope(final EtlMetrics metrics, final String scopeName, fina
             ourMetrics = metrics;
         }
 
-        if (rootMetrics.get() == null) {
-            rootMetrics.set(ourMetrics);
-            isRoot = true;
-        } else {
-            isRoot = false;
-        }
-
         this.scopeName = scopeName;
         metricsStart = System.currentTimeMillis();
     }
@@ -117,10 +108,6 @@ public void close() {
         }
         closed = true;
 
-        if (isRoot) {
-            rootMetrics.remove();
-        }
-
         if (ourMetrics != null) {
             final long metricsEnd = System.currentTimeMillis();
             ourMetrics.addTime(scopeName, metricsEnd - metricsStart);

diff --git a/src/main/java/com/amazon/pocketEtl/EtlTransformStage.java b/src/main/java/com/amazon/pocketEtl/EtlTransformStage.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -16,21 +16,21 @@
 package com.amazon.pocketEtl;
 
 
+import static org.apache.logging.log4j.LogManager.getLogger;
+
+import java.util.function.Function;
+
+import javax.annotation.Nonnull;
+
 import com.amazon.pocketEtl.core.consumer.EtlConsumer;
 import com.amazon.pocketEtl.core.consumer.EtlConsumerFactory;
 import com.amazon.pocketEtl.core.executor.EtlExecutor;
 import com.amazon.pocketEtl.core.executor.EtlExecutorFactory;
+
 import lombok.AccessLevel;
-import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
-import javax.annotation.Nonnull;
-import java.util.function.Function;
-
-import static org.apache.logging.log4j.LogManager.getLogger;
-
 @Getter(AccessLevel.PACKAGE)
-@EqualsAndHashCode(callSuper = true)
 class EtlTransformStage<T> extends EtlConsumerStage<T> {
     private final static String DEFAULT_TRANSFORM_STAGE_NAME = "EtlStream.Transform";
 

diff --git a/src/main/java/com/amazon/pocketEtl/Extractor.java b/src/main/java/com/amazon/pocketEtl/Extractor.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -17,7 +17,8 @@
 
 import javax.annotation.Nullable;
 import java.util.Optional;
-import java.util.prefs.BackingStoreException;
+
+import com.amazon.pocketEtl.exception.UnrecoverableStreamFailureException;
 
 /**
  * Interface for an extractor object that either extracts objects from some kind of storage.
@@ -31,9 +32,10 @@ public interface Extractor<T> extends AutoCloseable {
      *
      * @return An optional object that contains the next extracted object or empty if there are no objects left to be
      * extracted.
-     * @throws BackingStoreException If there was a problem extracting the object from the backing store.
+     * @throws UnrecoverableStreamFailureException An unrecoverable problem that affects the entire stream has been
+     * detected and the stream needs to be aborted.
      */
-    Optional<T> next() throws BackingStoreException;
+    Optional<T> next() throws UnrecoverableStreamFailureException;
 
     /**
      * Signal the extractor to prepare to extract objects.

diff --git a/src/main/java/com/amazon/pocketEtl/Loader.java b/src/main/java/com/amazon/pocketEtl/Loader.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -17,6 +17,8 @@
 
 import javax.annotation.Nullable;
 
+import com.amazon.pocketEtl.exception.UnrecoverableStreamFailureException;
+
 /**
  * Interface for a Loader that loads (writes) objects into final its final destination.
  *
@@ -28,8 +30,10 @@ public interface Loader<T> extends AutoCloseable {
      * Load a single object to to the destination store/service.
      *
      * @param objectToLoad The object to be loaded.
+     * @throws UnrecoverableStreamFailureException An unrecoverable problem that affects the entire stream has been
+     * detected and the stream needs to be aborted.
      */
-    void load(T objectToLoad);
+    void load(T objectToLoad) throws UnrecoverableStreamFailureException;
 
     /**
      * Signal the loader to prepare to load objects.

diff --git a/src/main/java/com/amazon/pocketEtl/Transformer.java b/src/main/java/com/amazon/pocketEtl/Transformer.java
@@ -1,5 +1,5 @@
 /*
- *   Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *   Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  *   Licensed under the Apache License, Version 2.0 (the "License").
  *   You may not use this file except in compliance with the License.
@@ -18,6 +18,8 @@
 import javax.annotation.Nullable;
 import java.util.List;
 
+import com.amazon.pocketEtl.exception.UnrecoverableStreamFailureException;
+
 /**
  * Interface for an object that transforms an object of one type into a stream of objects of another type. This allows
  * for an expansion or contraction of the stream where one object can become many objects or no objects.
@@ -32,8 +34,10 @@ public interface Transformer<UpstreamType, DownstreamType> extends AutoCloseable
      *
      * @param objectToTransform The object to be transformed.
      * @return The transformed object.
+     * @throws UnrecoverableStreamFailureException An unrecoverable problem that affects the entire stream has been
+     * detected and the stream needs to be aborted.
      */
-    List<DownstreamType> transform(UpstreamType objectToTransform);
+    List<DownstreamType> transform(UpstreamType objectToTransform) throws UnrecoverableStreamFailureException;
 
     /**
      * Signal the transformer to prepare to transform objects.