composite iterator

airbyte-commons/src/main/java/io/airbyte/commons/util/AutoCloseableIterators.java

@@ -24,22 +24,22 @@
 
 package io.airbyte.commons.util;
 
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterators;
 import io.airbyte.commons.concurrency.VoidCallable;
 import io.airbyte.commons.concurrency.VoidCallableNoException;
-import java.util.Collections;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class AutoCloseableIterators {
 
-  public static <T> AutoCloseableIterator<T> emptyIterator() {
-    return new DefaultAutoCloseableIterator<>(Collections.emptyIterator(), VoidCallable.NOOP);
-  }
+  private static final Logger LOGGER = LoggerFactory.getLogger(AutoCloseableIterators.class);
 
   /**
    * Coerces a vanilla {@link Iterator} into a {@link AutoCloseableIterator} by adding a no op close
@@ -102,12 +102,12 @@ public static <T> AutoCloseableIterator<T> lazyIterator(Supplier<AutoCloseableIt
    * @return autocloseable iterator
    */
   public static <T> AutoCloseableIterator<T> decorateWithEagerClose(AutoCloseableIterator<T> autoCloseableIterator) {
-    return new AutoCloseIterator<>(autoCloseableIterator);
+    return new EagerCloseIterator<>(autoCloseableIterator);
   }
 
   /**
    * Returns a {@link AutoCloseableIterator} that is composed of a {@link DefaultAutoCloseableIterator},
-   * {@link LazyAutoCloseableIterator}, and {@link AutoCloseIterator}.
+   * {@link LazyAutoCloseableIterator}, and {@link EagerCloseIterator}.
    *
    * @param streamSupplier supplies the stream this supplier will be called one time.
    * @param <T> type
@@ -172,15 +172,47 @@ public static <T> AutoCloseableIterator<T> transform(Function<AutoCloseableItera
    * @return concatenated iterator
    */
   public static <T> AutoCloseableIterator<T> concat(List<AutoCloseableIterator<T>> iterators) {
-    return new DefaultAutoCloseableIterator<>(Iterators.concat(iterators.iterator()), () -> {
-      for (AutoCloseableIterator<T> iterator : iterators) {
+    return new DefaultAutoCloseableIterator<>(Iterators.concat(iterators.iterator()), () -> iteratorCloseFunction(iterators));
+  }
+
+  public static <T> CompositeIterator<T> concatWithEagerClose(List<AutoCloseableIterator<T>> iterators) {
+    return DefaultCompositeIterator.fromAutoCloseableIterators(iterators);
+  }
+
+  public static class DefaultCompositeIterator<T> extends DefaultAutoCloseableIterator<T> implements CompositeIterator<T> {
+
+    private DefaultCompositeIterator(AutoCloseableIterator<T> iterator, VoidCallable onClose) {
+      super(iterator, onClose);
+    }
+
+    public static <T> CompositeIterator<T> fromAutoCloseableIterators(List<AutoCloseableIterator<T>> iterators) {
+      return new DefaultCompositeIterator<>(concatWithEagerClose(iterators), () -> iteratorCloseFunction(iterators));
+    }
+
+    public static <T> AutoCloseableIterator<T> concatWithEagerClose(List<AutoCloseableIterator<T>> iterators) {
+      final List<AutoCloseableIterator<T>> eagerCloseIterators = iterators
+          .stream()
+          .map(EagerCloseIterator::new)
+          .collect(Collectors.toList());
+
+      return AutoCloseableIterators.concat(eagerCloseIterators);
+    }
+  }
+
+  private static <T> void iteratorCloseFunction(List<AutoCloseableIterator<T>> iterators) throws Exception {
+    final List<Exception> exceptions = new ArrayList<>();
+    for (AutoCloseableIterator<T> iterator : iterators) {
+      try {
         iterator.close();
+      } catch (Exception e) {
+        LOGGER.error("exception while closing", e);
+        exceptions.add(e);
       }
-    });
-  }
+    }
 
-  public static <T> AutoCloseableIterator<T> concat(AutoCloseableIterator<T> iterator1, AutoCloseableIterator<T> iterator2) {
-    return concat(ImmutableList.of(iterator1, iterator2));
+    if (!exceptions.isEmpty()) {
+      throw exceptions.get(0);
+    }
   }
 
 }

airbyte-commons/src/main/java/io/airbyte/commons/util/CompositeIterator.java

@@ -0,0 +1,27 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.commons.util;
+
+public interface CompositeIterator<T> extends AutoCloseableIterator<T> {}

airbyte-commons/src/main/java/io/airbyte/commons/util/AutoCloseIterator.java → airbyte-commons/src/main/java/io/airbyte/commons/util/EagerCloseIterator.java

@@ -34,14 +34,14 @@
  *
  * @param <T> type
  */
-class AutoCloseIterator<T> extends AbstractIterator<T> implements AutoCloseableIterator<T> {
+class EagerCloseIterator<T> extends AbstractIterator<T> implements AutoCloseableIterator<T> {
 
   private final AutoCloseableIterator<T> internalIterator;
   private final VoidCallableNoException onClose;
 
   private boolean hasClosed;
 
-  public AutoCloseIterator(AutoCloseableIterator<T> iterator) {
+  public EagerCloseIterator(AutoCloseableIterator<T> iterator) {
     this.internalIterator = iterator;
     this.onClose = VoidCallableNoException.fromVoidCallable(iterator::close);
   }

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -57,6 +57,7 @@
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -162,7 +163,7 @@ public AutoCloseableIterator<AirbyteMessage> read(JsonNode config, ConfiguredAir
             .stream()
             .collect(Collectors.toMap(t -> String.format("%s.%s", t.getSchemaName(), t.getName()), Function.identity()));
 
-    AutoCloseableIterator<AirbyteMessage> resultIterator = AutoCloseableIterators.emptyIterator();
+    final List<AutoCloseableIterator<AirbyteMessage>> iteratorList = new ArrayList<>();
 
     for (final ConfiguredAirbyteStream airbyteStream : catalog.getStreams()) {
       final String streamName = airbyteStream.getStream().getName();
@@ -178,10 +179,10 @@ public AutoCloseableIterator<AirbyteMessage> read(JsonNode config, ConfiguredAir
           table,
           stateManager,
           emittedAt);
-      resultIterator = AutoCloseableIterators.concat(resultIterator, tableReadIterator);
+      iteratorList.add(tableReadIterator);
     }
 
-    return AutoCloseableIterators.appendOnClose(resultIterator, () -> {
+    return AutoCloseableIterators.appendOnClose(AutoCloseableIterators.concatWithEagerClose(iteratorList), () -> {
       LOGGER.info("Closing database connection pool.");
       Exceptions.toRuntime(database::close);
       LOGGER.info("Closed database connection pool.");