fix-baselibrary

Issue-hunters · Oct 16, 2023 · a3963fe · a3963fe
1 parent c5c48c1
commit a3963fe
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 6 deletions.
diff --git a/...-opensearch/src/main/java/io/airbyte/integrations/source/opensearch/OpenSearchSource.java b/...-opensearch/src/main/java/io/airbyte/integrations/source/opensearch/OpenSearchSource.java
@@ -8,11 +8,11 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.airbyte.cdk.integrations.BaseConnector;
+import io.airbyte.cdk.integrations.base.IntegrationRunner;
+import io.airbyte.cdk.integrations.base.Source;
 import io.airbyte.commons.util.AutoCloseableIterator;
 import io.airbyte.commons.util.AutoCloseableIterators;
-import io.airbyte.integrations.BaseConnector;
-import io.airbyte.integrations.base.IntegrationRunner;
-import io.airbyte.integrations.base.Source;
 import io.airbyte.protocol.models.v0.AirbyteCatalog;
 import io.airbyte.protocol.models.v0.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.v0.AirbyteMessage;

diff --git a/...e-opensearch/src/main/java/io/airbyte/integrations/source/opensearch/OpenSearchUtils.java b/...e-opensearch/src/main/java/io/airbyte/integrations/source/opensearch/OpenSearchUtils.java
@@ -5,28 +5,31 @@
 package io.airbyte.integrations.source.opensearch;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.stream.AirbyteStreamUtils;
 import io.airbyte.commons.util.AutoCloseableIterator;
 import io.airbyte.commons.util.AutoCloseableIterators;
+import io.airbyte.protocol.models.AirbyteStreamNameNamespacePair;
 import io.airbyte.protocol.models.v0.AirbyteMessage;
 import io.airbyte.protocol.models.v0.AirbyteRecordMessage;
 import io.airbyte.protocol.models.v0.AirbyteStream;
 import java.time.Instant;
 import java.util.List;
 
 public class OpenSearchUtils {
-
   public static AutoCloseableIterator<JsonNode> getDataIterator(final OpenSearchConnection connection,
                                                                 final AirbyteStream stream) {
+    final AirbyteStreamNameNamespacePair airbyteStream = AirbyteStreamUtils.convertFromAirbyteStream(stream);
     return AutoCloseableIterators.lazyIterator(() -> {
       try {
         List<JsonNode> data = connection.getRecords(stream.getName());
-        return AutoCloseableIterators.fromIterator(data.iterator());
+        return AutoCloseableIterators.fromIterator(data.iterator(), airbyteStream);
       } catch (final Exception e) {
         throw new RuntimeException(e);
       }
-    });
+    }, airbyteStream);
   }
 
+
   public static AutoCloseableIterator<AirbyteMessage> getMessageIterator(final AutoCloseableIterator<JsonNode> recordIterator,
                                                                          final String streamName) {
     return AutoCloseableIterators.transform(recordIterator, r -> new AirbyteMessage()