improve should use cdc check

Brigad · Jun 22, 2022 · 7d01727 · 7d01727
1 parent 24d7fa8
commit 7d01727
Show file tree

Hide file tree

Showing 6 changed files with 17 additions and 20 deletions.
diff --git a/.python-version b/.python-version
diff --git a/...ebezium-v1-4-2/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java b/...ebezium-v1-4-2/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java
@@ -17,6 +17,8 @@
 import io.airbyte.integrations.debezium.internals.FilteredFileDatabaseHistory;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import io.airbyte.protocol.models.ConfiguredAirbyteStream;
+import io.airbyte.protocol.models.SyncMode;
 import io.debezium.engine.ChangeEvent;
 import java.time.Instant;
 import java.util.Collections;
@@ -120,4 +122,9 @@ private Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager(final CdcSave
     return Optional.empty();
   }
 
+  public static boolean shouldUseCDC(final ConfiguredAirbyteCatalog catalog) {
+    return catalog.getStreams().stream().map(ConfiguredAirbyteStream::getSyncMode)
+        .anyMatch(syncMode -> syncMode == SyncMode.INCREMENTAL);
+  }
+
 }
diff --git a/...ebezium-v1-9-2/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java b/...ebezium-v1-9-2/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java
@@ -17,6 +17,8 @@
 import io.airbyte.integrations.debezium.internals.FilteredFileDatabaseHistory;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import io.airbyte.protocol.models.ConfiguredAirbyteStream;
+import io.airbyte.protocol.models.SyncMode;
 import io.debezium.engine.ChangeEvent;
 import java.time.Instant;
 import java.util.Collections;
@@ -120,4 +122,9 @@ private Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager(final CdcSave
     return Optional.empty();
   }
 
+  public static boolean shouldUseCDC(final ConfiguredAirbyteCatalog catalog) {
+    return catalog.getStreams().stream().map(ConfiguredAirbyteStream::getSyncMode)
+        .anyMatch(syncMode -> syncMode == SyncMode.INCREMENTAL);
+  }
+
 }
diff --git a/...nnectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java b/...nnectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java
@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.source.mssql;
 
+import static io.airbyte.integrations.debezium.AirbyteDebeziumHandler.shouldUseCDC;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_DELETED_AT;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_UPDATED_AT;
 import static java.util.stream.Collectors.toList;
@@ -382,13 +383,6 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(
     }
   }
 
-  private static boolean shouldUseCDC(final ConfiguredAirbyteCatalog catalog) {
-    final Optional<SyncMode> any = catalog.getStreams().stream()
-        .map(ConfiguredAirbyteStream::getSyncMode)
-        .filter(syncMode -> syncMode == SyncMode.INCREMENTAL).findAny();
-    return any.isPresent();
-  }
-
   // Note: in place mutation.
   private static AirbyteStream removeIncrementalWithoutPk(final AirbyteStream stream) {
     if (stream.getSourceDefinedPrimaryKey().isEmpty()) {

diff --git a/...nnectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java b/...nnectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java
@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.source.mysql;
 
+import static io.airbyte.integrations.debezium.AirbyteDebeziumHandler.shouldUseCDC;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_DELETED_AT;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_UPDATED_AT;
 import static io.airbyte.integrations.source.mysql.helpers.CdcConfigurationHelper.checkBinlog;
@@ -168,12 +169,6 @@ private static boolean isCdc(final JsonNode config) {
             .equals(ReplicationMethod.CDC);
   }
 
-  private static boolean shouldUseCDC(final ConfiguredAirbyteCatalog catalog) {
-    final Optional<SyncMode> any = catalog.getStreams().stream().map(ConfiguredAirbyteStream::getSyncMode)
-        .filter(syncMode -> syncMode == SyncMode.INCREMENTAL).findAny();
-    return any.isPresent();
-  }
-
   @Override
   public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final JdbcDatabase database,
                                                                              final ConfiguredAirbyteCatalog catalog,

diff --git a/...source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java b/...source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java
@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.source.postgres;
 
+import static io.airbyte.integrations.debezium.AirbyteDebeziumHandler.shouldUseCDC;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_DELETED_AT;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_UPDATED_AT;
 import static java.util.stream.Collectors.toList;
@@ -249,12 +250,6 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(
     }
   }
 
-  private static boolean shouldUseCDC(final ConfiguredAirbyteCatalog catalog) {
-    final Optional<SyncMode> any = catalog.getStreams().stream().map(ConfiguredAirbyteStream::getSyncMode)
-        .filter(syncMode -> syncMode == SyncMode.INCREMENTAL).findAny();
-    return any.isPresent();
-  }
-
   @VisibleForTesting
   static boolean isCdc(final JsonNode config) {
     final boolean isCdc = config.hasNonNull("replication_method")